diff --git a/.changeset/workflow-trigger-http.md b/.changeset/workflow-trigger-http.md new file mode 100644 index 00000000..83a891f4 --- /dev/null +++ b/.changeset/workflow-trigger-http.md @@ -0,0 +1,5 @@ +--- +"cre-cli": minor +--- + +Add a hidden `cre workflow trigger http ` command that fires a deployed workflow's HTTP trigger through the CRE Gateway. It signs a `workflows.execute` JSON-RPC request with the `CRE_ETH_PRIVATE_KEY` wallet (which must be one of the workflow's authorized keys), resolves the workflow ID from the selected `--target` unless `--workflow-id` is given, accepts `--input` as inline JSON or `@file`, and prints a machine-readable result with `--json`. diff --git a/cmd/root_test.go b/cmd/root_test.go index 274c9ee7..e7dea6c8 100644 --- a/cmd/root_test.go +++ b/cmd/root_test.go @@ -1,10 +1,12 @@ package cmd import ( + "bytes" "testing" "github.com/spf13/cobra" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) // buildCommandPath constructs a minimal cobra command hierarchy so that @@ -53,3 +55,29 @@ func TestIsRegistryRPCCommand(t *testing.T) { }) } } + +func TestWorkflowTriggerHTTP_HiddenFromHelpButWired(t *testing.T) { + workflowCmd, _, err := RootCmd.Find([]string{"workflow"}) + require.NoError(t, err) + + var help bytes.Buffer + workflowCmd.SetOut(&help) + t.Cleanup(func() { workflowCmd.SetOut(nil) }) + require.NoError(t, workflowCmd.Help()) + assert.Contains(t, help.String(), "simulate", "workflow help should still render its commands") + assert.NotContains(t, help.String(), "trigger", "trigger must stay out of `cre workflow --help`") + + httpCmd, _, err := RootCmd.Find([]string{"workflow", "trigger", "http"}) + require.NoError(t, err) + assert.Equal(t, "cre workflow trigger http", httpCmd.CommandPath()) + assert.True(t, httpCmd.Hidden) + assert.True(t, httpCmd.Parent().Hidden) +} + +func TestWorkflowTriggerHTTP_NeedsLoginAndWorkflowSettings(t *testing.T) { + httpCmd := buildCommandPath("cre", "workflow", "trigger", "http") + + assert.True(t, isLoadCredentials(httpCmd), "workflow lookup and the tenant gateway URL need a logged-in session") + assert.True(t, isLoadSettings(httpCmd), "the target, workflow name and wallet key come from workflow settings") + assert.False(t, isRegistryRPCCommand(httpCmd), "triggering must not require a registry RPC URL") +} diff --git a/cmd/workflow/trigger/http/http.go b/cmd/workflow/trigger/http/http.go new file mode 100644 index 00000000..47150e3b --- /dev/null +++ b/cmd/workflow/trigger/http/http.go @@ -0,0 +1,172 @@ +package http + +import ( + "context" + "crypto/ecdsa" + "encoding/json" + "fmt" + "io" + nethttp "net/http" + "strings" + "time" + + "github.com/ethereum/go-ethereum/crypto" + "github.com/spf13/cobra" + + "github.com/smartcontractkit/cre-cli/cmd/secrets/common/gateway" + "github.com/smartcontractkit/cre-cli/internal/runtime" + "github.com/smartcontractkit/cre-cli/internal/settings" + "github.com/smartcontractkit/cre-cli/internal/ui" + "github.com/smartcontractkit/cre-cli/internal/workflowresolve" +) + +const ( + defaultInput = "{}" + requestTimeout = 90 * time.Second +) + +// Inputs holds resolved and validated flag values for workflow trigger http. +type Inputs struct { + Input json.RawMessage + WorkflowID string + OutputFormat string +} + +func resolveInputs(inputFlag, workflowIDFlag, outputFormat string, jsonFlag bool, invocationDir string) (Inputs, error) { + format, err := workflowresolve.ResolveOutputFormat(outputFormat, jsonFlag) + if err != nil { + return Inputs{}, err + } + + input, err := parseInput(inputFlag, invocationDir) + if err != nil { + return Inputs{}, err + } + + var workflowID string + if strings.TrimSpace(workflowIDFlag) != "" { + if workflowID, err = normalizeWorkflowID(workflowIDFlag); err != nil { + return Inputs{}, fmt.Errorf("--workflow-id: %w", err) + } + } + + return Inputs{Input: input, WorkflowID: workflowID, OutputFormat: format}, nil +} + +// Handler signs a `workflows.execute` request with the wallet key and posts it to the Gateway. +type Handler struct { + runtimeContext *runtime.Context + out io.Writer +} + +// NewHandler builds a Handler that writes machine-readable output to out. +func NewHandler(ctx *runtime.Context, out io.Writer) *Handler { + return &Handler{runtimeContext: ctx, out: out} +} + +// Execute fires the trigger and reports the Gateway outcome. It returns an error for any +// failure, after printing the JSON result when --json is set. +func (h *Handler) Execute(ctx context.Context, inputs Inputs) error { + rt := h.runtimeContext + if rt.Settings == nil { + return fmt.Errorf("workflow settings not loaded; ensure workflow.yaml is valid") + } + + key, err := h.signingKey() + if err != nil { + return err + } + + gatewayURL := gateway.ResolveVaultGatewayURL(rt.TenantContext, rt.EnvironmentSet) + if err := gateway.ValidateGatewayURL(gatewayURL); err != nil { + return err + } + + workflowID := inputs.WorkflowID + if workflowID == "" { + if workflowID, err = h.resolveWorkflowID(ctx); err != nil { + return err + } + } + rt.Workflow.ID = strings.TrimPrefix(workflowID, "0x") + + signed, err := newSignedRequest(key, workflowID, inputs.Input) + if err != nil { + return err + } + rt.Logger.Debug().Str("workflowID", workflowID).Str("requestID", signed.ID).Str("signer", signed.Signer.Hex()).Msg("Sending HTTP trigger request") + + // JWTs are single-use, so a retried POST would be rejected as a replay. + gw := &gateway.HTTPClient{URL: gatewayURL, Client: &nethttp.Client{Timeout: requestTimeout}, RetryAttempts: 1} + + spinner := ui.NewSpinner() + spinner.Start("Sending trigger request to the Gateway...") + respBody, _, postErr := gw.PostWithBearer(signed.Body, signed.Token) + spinner.Stop() + + result, rpcErr := parseGatewayResponse(respBody, postErr) + out := newOutput(workflowID, signed.ID, gatewayURL, result, rpcErr) + + if inputs.OutputFormat == workflowresolve.OutputFormatJSON { + if err := printJSON(h.out, out); err != nil { + return err + } + } else if rpcErr == nil { + printSummary(out) + } + + if rpcErr != nil { + return fmt.Errorf("trigger request %s failed: %s", signed.ID, rpcErr.Message) + } + return nil +} + +func (h *Handler) signingKey() (*ecdsa.PrivateKey, error) { + resolved, err := settings.ResolveEthPrivateKeyFromEnv(h.runtimeContext.Settings.User.PrivateKey(settings.EVM)) + if err != nil { + return nil, err + } + if !resolved.IsSet() { + return nil, fmt.Errorf("%s is not set; it must be one of the workflow's HTTP-trigger authorized keys", settings.EthPrivateKeyEnvVar) + } + + key, err := crypto.HexToECDSA(resolved.Hex()) + if err != nil { + return nil, fmt.Errorf("failed to decode the provided private key: %w", err) + } + return key, nil +} + +// New returns the cobra command. +func New(runtimeContext *runtime.Context) *cobra.Command { + var inputFlag, workflowIDFlag, outputFormat string + var jsonFlag bool + + cmd := &cobra.Command{ + Use: "http ", + Short: "Fires a deployed workflow's HTTP trigger through the CRE Gateway", + Hidden: true, // Kept hidden so the public CLI surface is unchanged until this is productized. + Long: `Signs a workflows.execute JSON-RPC request with the wallet key from ` + + `CRE_ETH_PRIVATE_KEY and posts it to the environment's Gateway. The key must be one of ` + + `the workflow's HTTP-trigger authorized keys. The deployed workflow ID is resolved from ` + + `the workflow name configured for the selected --target unless --workflow-id is given.`, + Example: `cre workflow trigger http ./my-workflow --target staging + cre workflow trigger http ./my-workflow --target staging --input '{"orderId": 42}' + cre workflow trigger http ./my-workflow --target staging --input @payload.json --json + cre workflow trigger http ./my-workflow --target staging --workflow-id 0x<64 hex chars>`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + inputs, err := resolveInputs(inputFlag, workflowIDFlag, outputFormat, jsonFlag, runtimeContext.InvocationDir) + if err != nil { + return err + } + return NewHandler(runtimeContext, cmd.OutOrStdout()).Execute(cmd.Context(), inputs) + }, + } + + cmd.Flags().StringVar(&inputFlag, "input", defaultInput, "Trigger input as a JSON object or array, or @ to read it from a file") + cmd.Flags().StringVar(&workflowIDFlag, "workflow-id", "", "Deployed workflow ID (0x...) to trigger instead of resolving it from the workflow folder") + cmd.Flags().StringVar(&outputFormat, "output", "", `Output format: "json" prints the result as JSON to stdout`) + cmd.Flags().BoolVar(&jsonFlag, "json", false, "Output as JSON (shorthand for --output=json)") + return cmd +} diff --git a/cmd/workflow/trigger/http/http_test.go b/cmd/workflow/trigger/http/http_test.go new file mode 100644 index 00000000..245750a6 --- /dev/null +++ b/cmd/workflow/trigger/http/http_test.go @@ -0,0 +1,570 @@ +package http_test + +import ( + "bytes" + "context" + "crypto/ecdsa" + "encoding/hex" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/ethereum/go-ethereum/crypto" + "github.com/google/uuid" + "github.com/rs/zerolog" + "github.com/spf13/cobra" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/pkg/jsonrpc2" + "github.com/smartcontractkit/chainlink/v2/core/utils" + + "github.com/smartcontractkit/cre-cli/cmd/workflow/trigger" + "github.com/smartcontractkit/cre-cli/internal/credentials" + "github.com/smartcontractkit/cre-cli/internal/environments" + "github.com/smartcontractkit/cre-cli/internal/runtime" + "github.com/smartcontractkit/cre-cli/internal/settings" + "github.com/smartcontractkit/cre-cli/internal/tenantctx" +) + +const ( + // Platform rows carry the workflow ID without a 0x prefix. + platformWorkflowID = "00da21b8b3e117e31f3a3e8a0795225cbde6c00283a84395117669691f2b7856" + wantWorkflowID = "0x" + platformWorkflowID +) + +type gatewayCall struct { + method string + contentType string + accept string + authorization string + body []byte +} + +type fakeGateway struct { + *httptest.Server + mu sync.Mutex + calls []gatewayCall +} + +func (g *fakeGateway) recorded() []gatewayCall { + g.mu.Lock() + defer g.mu.Unlock() + return append([]gatewayCall(nil), g.calls...) +} + +func newFakeGateway(t *testing.T, status int, reply func(requestID string) any) *fakeGateway { + t.Helper() + gw := &fakeGateway{} + gw.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + gw.mu.Lock() + gw.calls = append(gw.calls, gatewayCall{ + method: r.Method, + contentType: r.Header.Get("Content-Type"), + accept: r.Header.Get("Accept"), + authorization: r.Header.Get("Authorization"), + body: body, + }) + gw.mu.Unlock() + + var req jsonrpc2.Request[json.RawMessage] + _ = json.Unmarshal(body, &req) + w.Header().Set("Content-Type", "application/jsonrpc") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(reply(req.ID)) + })) + t.Cleanup(gw.Close) + return gw +} + +func acceptedReply(requestID string) any { + return map[string]any{ + "jsonrpc": "2.0", + "id": requestID, + "method": "workflows.execute", + "result": map[string]any{"workflow_id": platformWorkflowID, "workflow_execution_id": "0xe1", "status": "ACCEPTED"}, + } +} + +func rpcErrorReply(code int64, message string) func(string) any { + return func(requestID string) any { + return map[string]any{ + "jsonrpc": "2.0", + "id": requestID, + "error": map[string]any{"code": code, "message": message}, + } + } +} + +func newKey(t *testing.T) (*ecdsa.PrivateKey, string) { + t.Helper() + key, err := crypto.GenerateKey() + require.NoError(t, err) + return key, hex.EncodeToString(crypto.FromECDSA(key)) +} + +func newRuntimeContext(gatewayURL, privateKeyHex string) *runtime.Context { + logger := zerolog.Nop() + s := &settings.Settings{ + User: settings.UserSettings{ + TargetName: "staging", + PrivateKeys: map[string]string{settings.EVM.Name: privateKeyHex}, + }, + } + s.Workflow.UserWorkflowSettings.WorkflowName = "alpha" + + return &runtime.Context{ + Logger: &logger, + Settings: s, + Credentials: &credentials.Credentials{AuthType: credentials.AuthTypeApiKey, APIKey: "test-key"}, + EnvironmentSet: &environments.EnvironmentSet{GatewayURL: gatewayURL}, + TenantContext: &tenantctx.EnvironmentContext{}, + } +} + +// execute runs `cre workflow trigger http` under a minimal root that mirrors the real global flags. +func execute(t *testing.T, rtCtx *runtime.Context, args ...string) (stdout string, err error) { + t.Helper() + root := &cobra.Command{Use: "cre", SilenceUsage: true, SilenceErrors: true} + root.PersistentFlags().String("target", "", "") + root.PersistentFlags().Bool("non-interactive", false, "") + workflowCmd := &cobra.Command{Use: "workflow"} + workflowCmd.AddCommand(trigger.New(rtCtx)) + root.AddCommand(workflowCmd) + + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(io.Discard) + root.SetArgs(append([]string{"workflow", "trigger", "http", "./my-workflow", "--target", "staging", "--non-interactive"}, args...)) + + err = root.ExecuteContext(context.Background()) + return out.String(), err +} + +// captureProcessStdout collects what ui helpers print straight to os.Stdout. +func captureProcessStdout(t *testing.T, fn func()) string { + t.Helper() + r, w, err := os.Pipe() + require.NoError(t, err) + old := os.Stdout + os.Stdout = w + fn() + require.NoError(t, w.Close()) + os.Stdout = old + var buf strings.Builder + _, err = io.Copy(&buf, r) + require.NoError(t, err) + return buf.String() +} + +func decodeJSONOutput(t *testing.T, stdout string) map[string]json.RawMessage { + t.Helper() + var out map[string]json.RawMessage + require.NoError(t, json.Unmarshal([]byte(stdout), &out), "stdout must be a single JSON document, got: %q", stdout) + return out +} + +func TestTriggerHTTP_SignsAndPostsRequestAndPrintsJSON(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + key, keyHex := newKey(t) + rtCtx := newRuntimeContext(gw.URL, keyHex) + + var stdout string + processStdout := captureProcessStdout(t, func() { + var err error + // Upper-case, unprefixed ID checks that the CLI normalizes to what the Gateway accepts. + stdout, err = execute(t, rtCtx, "--workflow-id", strings.ToUpper(platformWorkflowID), "--input", `{"orderId":42}`, "--json") + require.NoError(t, err) + }) + assert.Empty(t, processStdout, "with --json nothing but the JSON document may reach stdout") + + calls := gw.recorded() + require.Len(t, calls, 1) + call := calls[0] + assert.Equal(t, http.MethodPost, call.method) + assert.Equal(t, "application/jsonrpc", call.contentType) + assert.Equal(t, "application/json", call.accept) + token, ok := strings.CutPrefix(call.authorization, "Bearer ") + require.True(t, ok, "Authorization must be a Bearer JWT, got %q", call.authorization) + + req, err := jsonrpc2.DecodeRequest[json.RawMessage](call.body, token) + require.NoError(t, err) + assert.Equal(t, "2.0", req.Version) + assert.Equal(t, "workflows.execute", req.Method) + _, parseErr := uuid.Parse(req.ID) + require.NoError(t, parseErr) + assert.JSONEq(t, `{"input":{"orderId":42},"workflow":{"workflowID":"`+wantWorkflowID+`"}}`, string(*req.Params)) + + _, signer, err := utils.VerifyRequestJWT(token, req) + require.NoError(t, err, "the Gateway must be able to verify the JWT against the body") + assert.Equal(t, crypto.PubkeyToAddress(key.PublicKey), signer) + + out := decodeJSONOutput(t, stdout) + assert.Len(t, out, 5, "unexpected keys in %s", stdout) + assert.JSONEq(t, `"`+wantWorkflowID+`"`, string(out["workflowId"])) + assert.JSONEq(t, `"`+req.ID+`"`, string(out["requestId"])) + assert.JSONEq(t, `"`+gw.URL+`"`, string(out["gatewayUrl"])) + assert.JSONEq(t, `"ACCEPTED"`, string(out["status"])) + assert.JSONEq(t, `{"workflow_id":"`+platformWorkflowID+`","workflow_execution_id":"0xe1","status":"ACCEPTED"}`, string(out["result"])) +} + +func TestTriggerHTTP_OutputJSONFlagMatchesJSONShorthand(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + _, keyHex := newKey(t) + + stdout, err := execute(t, newRuntimeContext(gw.URL, keyHex), "--workflow-id", wantWorkflowID, "--output", "json") + + require.NoError(t, err) + assert.JSONEq(t, `"ACCEPTED"`, string(decodeJSONOutput(t, stdout)["status"])) +} + +func TestTriggerHTTP_RejectsUnsupportedOutputFormat(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + _, keyHex := newKey(t) + + _, err := execute(t, newRuntimeContext(gw.URL, keyHex), "--workflow-id", wantWorkflowID, "--output", "yaml") + + require.Error(t, err) + assert.Contains(t, err.Error(), "only \"json\" is accepted") + assert.Empty(t, gw.recorded()) +} + +func TestTriggerHTTP_DefaultInputIsEmptyObject(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + _, keyHex := newKey(t) + + _, err := execute(t, newRuntimeContext(gw.URL, keyHex), "--workflow-id", wantWorkflowID, "--json") + + require.NoError(t, err) + calls := gw.recorded() + require.Len(t, calls, 1) + var body struct { + Params struct { + Input json.RawMessage `json:"input"` + } `json:"params"` + } + require.NoError(t, json.Unmarshal(calls[0].body, &body)) + assert.JSONEq(t, `{}`, string(body.Params.Input)) +} + +func TestTriggerHTTP_JSONRPCErrorPrintsErrorAndFails(t *testing.T) { + gw := newFakeGateway(t, http.StatusBadRequest, rpcErrorReply(jsonrpc2.ErrInvalidRequest, "Auth failure: signer is not authorized")) + _, keyHex := newKey(t) + + stdout, err := execute(t, newRuntimeContext(gw.URL, keyHex), "--workflow-id", wantWorkflowID, "--json") + + require.Error(t, err) + assert.Contains(t, err.Error(), "Auth failure: signer is not authorized") + + out := decodeJSONOutput(t, stdout) + assert.JSONEq(t, `"ERROR"`, string(out["status"])) + assert.JSONEq(t, `null`, string(out["result"])) + assert.JSONEq(t, `{"code":-32600,"message":"Auth failure: signer is not authorized"}`, string(out["error"])) + assert.JSONEq(t, `"`+wantWorkflowID+`"`, string(out["workflowId"])) + assert.NotEmpty(t, out["requestId"]) + assert.JSONEq(t, `"`+gw.URL+`"`, string(out["gatewayUrl"])) +} + +func TestTriggerHTTP_JSONRPCErrorOnHTTP200StillFails(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, rpcErrorReply(jsonrpc2.ErrInternal, "internal error")) + _, keyHex := newKey(t) + + stdout, err := execute(t, newRuntimeContext(gw.URL, keyHex), "--workflow-id", wantWorkflowID, "--json") + + require.Error(t, err) + assert.JSONEq(t, `{"code":-32603,"message":"internal error"}`, string(decodeJSONOutput(t, stdout)["error"])) +} + +func TestTriggerHTTP_TransportFailureFailsWithJSONError(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + _, keyHex := newKey(t) + rtCtx := newRuntimeContext(gw.URL, keyHex) + gw.Close() + + stdout, err := execute(t, rtCtx, "--workflow-id", wantWorkflowID, "--json") + + require.Error(t, err) + out := decodeJSONOutput(t, stdout) + assert.JSONEq(t, `"ERROR"`, string(out["status"])) + assert.Contains(t, string(out["error"]), "gateway") +} + +func TestTriggerHTTP_ReadsInputFromFileRelativeToInvocationDir(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + _, keyHex := newKey(t) + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "payload.json"), []byte("{\"from\": \"file\", \"n\": [1, 2]}\n"), 0o600)) + rtCtx := newRuntimeContext(gw.URL, keyHex) + rtCtx.InvocationDir = dir + + _, err := execute(t, rtCtx, "--workflow-id", wantWorkflowID, "--input", "@payload.json", "--json") + + require.NoError(t, err) + calls := gw.recorded() + require.Len(t, calls, 1) + var body struct { + Params struct { + Input json.RawMessage `json:"input"` + } `json:"params"` + } + require.NoError(t, json.Unmarshal(calls[0].body, &body)) + assert.JSONEq(t, `{"from":"file","n":[1,2]}`, string(body.Params.Input)) +} + +func TestTriggerHTTP_InvalidInputFailsBeforeAnyRequest(t *testing.T) { + tests := []struct { + name string + input string + wantErr string + }{ + {name: "malformed JSON", input: `{"orderId":`, wantErr: "--input is not valid JSON"}, + {name: "primitive", input: `42`, wantErr: "must be a JSON object or array"}, + {name: "missing file", input: "@does-not-exist.json", wantErr: "failed to read --input file"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + _, keyHex := newKey(t) + + stdout, err := execute(t, newRuntimeContext(gw.URL, keyHex), "--workflow-id", wantWorkflowID, "--input", tt.input, "--json") + + require.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErr) + assert.Empty(t, stdout) + assert.Empty(t, gw.recorded()) + }) + } +} + +func TestTriggerHTTP_RejectsMalformedWorkflowID(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + _, keyHex := newKey(t) + + _, err := execute(t, newRuntimeContext(gw.URL, keyHex), "--workflow-id", "0x1234") + + require.Error(t, err) + assert.Contains(t, err.Error(), "--workflow-id") + assert.Empty(t, gw.recorded()) +} + +func TestTriggerHTTP_RequiresWalletKey(t *testing.T) { + for name, key := range map[string]string{ + "unset": "", + "placeholder": settings.DefaultEthPrivateKeyEnvPlaceholder, + } { + t.Run(name, func(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + + _, err := execute(t, newRuntimeContext(gw.URL, key), "--workflow-id", wantWorkflowID) + + require.Error(t, err) + assert.Contains(t, err.Error(), "CRE_ETH_PRIVATE_KEY") + assert.Empty(t, gw.recorded()) + }) + } +} + +func TestTriggerHTTP_RejectsMalformedWalletKey(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + + _, err := execute(t, newRuntimeContext(gw.URL, "not-a-key"), "--workflow-id", wantWorkflowID) + + require.Error(t, err) + assert.Contains(t, err.Error(), "invalid private key") + assert.Empty(t, gw.recorded()) +} + +func TestTriggerHTTP_RejectsInsecureGatewayURL(t *testing.T) { + _, keyHex := newKey(t) + + _, err := execute(t, newRuntimeContext("http://gateway.example.com", keyHex), "--workflow-id", wantWorkflowID) + + require.Error(t, err) + assert.Contains(t, err.Error(), "https") +} + +func TestTriggerHTTP_PrintsFriendlySummaryWithoutJSONFlag(t *testing.T) { + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + _, keyHex := newKey(t) + + var cmdOut string + uiOut := captureProcessStdout(t, func() { + var err error + cmdOut, err = execute(t, newRuntimeContext(gw.URL, keyHex), "--workflow-id", wantWorkflowID) + require.NoError(t, err) + }) + + assert.Empty(t, cmdOut) + assert.Contains(t, uiOut, "Trigger request sent") + assert.Contains(t, uiOut, wantWorkflowID) + assert.Contains(t, uiOut, gw.URL) + assert.Contains(t, uiOut, "ACCEPTED") +} + +func TestTriggerHTTP_ErrorWithoutJSONFlagPrintsNothingToStdout(t *testing.T) { + gw := newFakeGateway(t, http.StatusBadRequest, rpcErrorReply(jsonrpc2.ErrInvalidRequest, "Workflow not found")) + _, keyHex := newKey(t) + + var cmdOut string + var err error + uiOut := captureProcessStdout(t, func() { + cmdOut, err = execute(t, newRuntimeContext(gw.URL, keyHex), "--workflow-id", wantWorkflowID) + }) + + require.Error(t, err) + assert.Contains(t, err.Error(), "Workflow not found") + assert.Empty(t, cmdOut) + assert.NotContains(t, uiOut, "Trigger request sent") +} + +// platformServer serves GraphQL at /graphql and the Gateway at /gateway. +func platformServer(t *testing.T, workflows []map[string]any, onGateway http.HandlerFunc) *httptest.Server { + t.Helper() + mux := http.NewServeMux() + mux.HandleFunc("/graphql", func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + query, _ := body["query"].(string) + require.Contains(t, query, "ListWorkflows") + + _ = json.NewEncoder(w).Encode(map[string]any{ + "data": map[string]any{"workflows": map[string]any{"count": len(workflows), "data": workflows}}, + }) + }) + mux.HandleFunc("/gateway", onGateway) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + return srv +} + +func workflowRow(name, id, status string) map[string]any { + return map[string]any{ + "uuid": "uuid-" + id[:4] + status, + "name": name, + "workflowId": id, + "ownerAddress": "2020202020202020202020202020202020202020", + "status": status, + "workflowSource": "private", + } +} + +func TestTriggerHTTP_ResolvesActiveWorkflowIDFromWorkflowName(t *testing.T) { + const olderID = "9999999999999999999999999999999999999999999999999999999999999999" + var gotID string + srv := platformServer(t, []map[string]any{ + workflowRow("alpha", olderID, "DELETED"), + workflowRow("alpha-suffix", olderID, "ACTIVE"), + workflowRow("Alpha", platformWorkflowID, "ACTIVE"), + }, func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + var req struct { + ID string `json:"id"` + Params struct { + Workflow struct { + WorkflowID string `json:"workflowID"` + } `json:"workflow"` + } `json:"params"` + } + _ = json.Unmarshal(body, &req) + gotID = req.Params.Workflow.WorkflowID + _ = json.NewEncoder(w).Encode(acceptedReply(req.ID)) + }) + _, keyHex := newKey(t) + rtCtx := newRuntimeContext(srv.URL+"/gateway", keyHex) + rtCtx.EnvironmentSet.GraphQLURL = srv.URL + "/graphql" + + stdout, err := execute(t, rtCtx, "--json") + + require.NoError(t, err) + assert.Equal(t, wantWorkflowID, gotID) + assert.JSONEq(t, `"`+wantWorkflowID+`"`, string(decodeJSONOutput(t, stdout)["workflowId"])) + assert.Equal(t, platformWorkflowID, rtCtx.Workflow.ID, "telemetry expects the ID without a 0x prefix") +} + +func TestTriggerHTTP_FailsWhenWorkflowNameHasNoMatch(t *testing.T) { + gatewayHit := false + srv := platformServer(t, []map[string]any{ + workflowRow("beta", platformWorkflowID, "ACTIVE"), + workflowRow("alpha", platformWorkflowID, "DELETED"), + }, func(http.ResponseWriter, *http.Request) { gatewayHit = true }) + _, keyHex := newKey(t) + rtCtx := newRuntimeContext(srv.URL+"/gateway", keyHex) + rtCtx.EnvironmentSet.GraphQLURL = srv.URL + "/graphql" + + _, err := execute(t, rtCtx, "--json") + + require.Error(t, err) + assert.Contains(t, err.Error(), `no workflow found with name "alpha"`) + assert.False(t, gatewayHit) +} + +func TestTriggerHTTP_FailsOnAmbiguousActiveWorkflows(t *testing.T) { + srv := platformServer(t, []map[string]any{ + workflowRow("alpha", platformWorkflowID, "ACTIVE"), + workflowRow("alpha", "7777777777777777777777777777777777777777777777777777777777777777", "ACTIVE"), + }, func(http.ResponseWriter, *http.Request) { t.Error("gateway must not be called") }) + _, keyHex := newKey(t) + rtCtx := newRuntimeContext(srv.URL+"/gateway", keyHex) + rtCtx.EnvironmentSet.GraphQLURL = srv.URL + "/graphql" + + _, err := execute(t, rtCtx) + + require.Error(t, err) + assert.Contains(t, err.Error(), "multiple ACTIVE workflows") + assert.Contains(t, err.Error(), "--workflow-id") +} + +func TestTriggerHTTP_WorkflowIDFlagSkipsPlatformLookup(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/graphql" { + t.Error("platform must not be queried when --workflow-id is given") + } + body, _ := io.ReadAll(r.Body) + var req jsonrpc2.Request[json.RawMessage] + _ = json.Unmarshal(body, &req) + _ = json.NewEncoder(w).Encode(acceptedReply(req.ID)) + })) + t.Cleanup(srv.Close) + _, keyHex := newKey(t) + rtCtx := newRuntimeContext(srv.URL+"/gateway", keyHex) + rtCtx.EnvironmentSet.GraphQLURL = srv.URL + "/graphql" + + _, err := execute(t, rtCtx, "--workflow-id", wantWorkflowID, "--json") + + require.NoError(t, err) +} + +func TestTriggerHTTP_ResolutionRequiresUserContextAndWorkflowName(t *testing.T) { + _, keyHex := newKey(t) + gw := newFakeGateway(t, http.StatusOK, acceptedReply) + + t.Run("no user context", func(t *testing.T) { + rtCtx := newRuntimeContext(gw.URL, keyHex) + rtCtx.TenantContext = nil + + _, err := execute(t, rtCtx) + + require.Error(t, err) + assert.Contains(t, err.Error(), "user context not available") + }) + + t.Run("no workflow name", func(t *testing.T) { + rtCtx := newRuntimeContext(gw.URL, keyHex) + rtCtx.Settings.Workflow.UserWorkflowSettings.WorkflowName = "" + + _, err := execute(t, rtCtx) + + require.Error(t, err) + assert.Contains(t, err.Error(), "workflow-name is not set") + }) + + assert.Empty(t, gw.recorded()) +} diff --git a/cmd/workflow/trigger/http/input.go b/cmd/workflow/trigger/http/input.go new file mode 100644 index 00000000..6bdf2d69 --- /dev/null +++ b/cmd/workflow/trigger/http/input.go @@ -0,0 +1,73 @@ +package http + +import ( + "bytes" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/smartcontractkit/cre-cli/internal/workflowresolve" +) + +// parseInput accepts inline JSON or @ and returns a validated JSON object or array. +func parseInput(raw, invocationDir string) (json.RawMessage, error) { + raw = strings.TrimSpace(raw) + source := "--input" + + if path, ok := strings.CutPrefix(raw, "@"); ok { + resolved := resolveFromInvocation(strings.TrimSpace(path), invocationDir) + data, err := os.ReadFile(resolved) // #nosec G304 -- path is an explicit CLI argument + if err != nil { + return nil, fmt.Errorf("failed to read --input file: %w", err) + } + raw = string(bytes.TrimSpace(data)) + source = fmt.Sprintf("--input file %s", resolved) + } + + var decoded any + if err := json.Unmarshal([]byte(raw), &decoded); err != nil { + return nil, fmt.Errorf("%s is not valid JSON: %w", source, err) + } + // The Gateway rejects primitives, so fail before spending a request on it. + switch decoded.(type) { + case map[string]any, []any: + default: + return nil, fmt.Errorf("%s must be a JSON object or array, got %s", source, jsonKind(decoded)) + } + + return json.RawMessage(raw), nil +} + +// resolveFromInvocation anchors relative paths at the directory the user ran the CLI from, +// because the process cwd is moved into the workflow folder before handlers run. +func resolveFromInvocation(path, invocationDir string) string { + if filepath.IsAbs(path) || invocationDir == "" { + return path + } + return filepath.Join(invocationDir, path) +} + +func jsonKind(v any) string { + switch v.(type) { + case nil: + return "null" + case string: + return "a string" + case bool: + return "a boolean" + default: + return "a number" + } +} + +// normalizeWorkflowID returns the 0x-prefixed lower-case form, which is the only casing the Gateway accepts. +func normalizeWorkflowID(raw string) (string, error) { + id := strings.TrimSpace(raw) + id = strings.TrimPrefix(strings.TrimPrefix(id, "0x"), "0X") + if !workflowresolve.LooksLikeWorkflowID(id) { + return "", fmt.Errorf("invalid workflow ID %q: expected 64 hex characters with an optional 0x prefix", raw) + } + return "0x" + strings.ToLower(id), nil +} diff --git a/cmd/workflow/trigger/http/input_test.go b/cmd/workflow/trigger/http/input_test.go new file mode 100644 index 00000000..90f94477 --- /dev/null +++ b/cmd/workflow/trigger/http/input_test.go @@ -0,0 +1,82 @@ +package http + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseInput(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "payload.json"), []byte("\n {\"from\": \"file\"}\n"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "scalar.json"), []byte(`"just a string"`), 0o600)) + + tests := []struct { + name string + raw string + invDir string + want string + wantErr string + }{ + {name: "empty object", raw: "{}", want: "{}"}, + {name: "object", raw: ` {"a": 1} `, want: `{"a": 1}`}, + {name: "array", raw: `[1, 2]`, want: `[1, 2]`}, + {name: "relative file resolves against invocation dir", raw: "@payload.json", invDir: dir, want: `{"from": "file"}`}, + {name: "absolute file", raw: "@" + filepath.Join(dir, "payload.json"), want: `{"from": "file"}`}, + {name: "invalid JSON", raw: `{oops`, wantErr: "--input is not valid JSON"}, + {name: "empty", raw: " ", wantErr: "--input is not valid JSON"}, + {name: "string primitive", raw: `"x"`, wantErr: "must be a JSON object or array, got a string"}, + {name: "number primitive", raw: `7`, wantErr: "must be a JSON object or array, got a number"}, + {name: "boolean primitive", raw: `true`, wantErr: "must be a JSON object or array, got a boolean"}, + {name: "null", raw: `null`, wantErr: "must be a JSON object or array, got null"}, + {name: "missing file", raw: "@nope.json", invDir: dir, wantErr: "failed to read --input file"}, + {name: "file with primitive", raw: "@scalar.json", invDir: dir, wantErr: "must be a JSON object or array"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := parseInput(tt.raw, tt.invDir) + if tt.wantErr != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErr) + return + } + require.NoError(t, err) + assert.Equal(t, tt.want, string(got)) + }) + } +} + +func TestNormalizeWorkflowID(t *testing.T) { + const hexID = "00da21b8b3e117e31f3a3e8a0795225cbde6c00283a84395117669691f2b7856" + + tests := []struct { + name string + raw string + want string + wantErr bool + }{ + {name: "bare hex gains prefix", raw: hexID, want: "0x" + hexID}, + {name: "prefixed hex", raw: "0x" + hexID, want: "0x" + hexID}, + {name: "upper case is lowered", raw: "0X00DA21B8B3E117E31F3A3E8A0795225CBDE6C00283A84395117669691F2B7856", want: "0x" + hexID}, + {name: "whitespace is trimmed", raw: " " + hexID + "\n", want: "0x" + hexID}, + {name: "too short", raw: "0x1234", wantErr: true}, + {name: "not hex", raw: "0x" + hexID[:62] + "zz", wantErr: true}, + {name: "empty", raw: "", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := normalizeWorkflowID(tt.raw) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/cmd/workflow/trigger/http/request.go b/cmd/workflow/trigger/http/request.go new file mode 100644 index 00000000..60884597 --- /dev/null +++ b/cmd/workflow/trigger/http/request.go @@ -0,0 +1,61 @@ +package http + +import ( + "crypto/ecdsa" + "encoding/json" + "fmt" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "github.com/google/uuid" + + "github.com/smartcontractkit/chainlink-common/pkg/jsonrpc2" + gwtypes "github.com/smartcontractkit/chainlink-common/pkg/types/gateway" + "github.com/smartcontractkit/chainlink/v2/core/utils" +) + +// triggerParams is gateway.HTTPTriggerRequest without the server-assigned key. It keeps +// input bytes verbatim, whereas that type's MarshalJSON round-trips them through float64. +type triggerParams struct { + Input json.RawMessage `json:"input"` + Workflow gwtypes.WorkflowSelector `json:"workflow"` +} + +// signedRequest is a `workflows.execute` JSON-RPC body plus the JWT that authorizes it. +type signedRequest struct { + ID string + Body []byte + Token string + Signer common.Address +} + +// newSignedRequest signs with the node's own JWT helper so the digest and ETH signature +// scheme cannot drift from what the Gateway verifies. +func newSignedRequest(key *ecdsa.PrivateKey, workflowID string, input json.RawMessage) (*signedRequest, error) { + req := jsonrpc2.Request[triggerParams]{ + Version: jsonrpc2.JsonRpcVersion, + ID: uuid.NewString(), + Method: gwtypes.MethodWorkflowExecute, + Params: &triggerParams{ + Input: input, + Workflow: gwtypes.WorkflowSelector{WorkflowID: workflowID}, + }, + } + + signer := crypto.PubkeyToAddress(key.PublicKey) + unsigned, err := utils.CreateRequestJWT(req, utils.WithIssuer(signer.Hex())) + if err != nil { + return nil, fmt.Errorf("failed to build request JWT: %w", err) + } + token, err := unsigned.SignedString(key) + if err != nil { + return nil, fmt.Errorf("failed to sign request JWT: %w", err) + } + + body, err := json.Marshal(req) + if err != nil { + return nil, fmt.Errorf("failed to encode trigger request: %w", err) + } + + return &signedRequest{ID: req.ID, Body: body, Token: token, Signer: signer}, nil +} diff --git a/cmd/workflow/trigger/http/request_test.go b/cmd/workflow/trigger/http/request_test.go new file mode 100644 index 00000000..dad92579 --- /dev/null +++ b/cmd/workflow/trigger/http/request_test.go @@ -0,0 +1,137 @@ +package http + +import ( + "crypto/ecdsa" + "encoding/base64" + "encoding/json" + "strings" + "testing" + "time" + + "github.com/ethereum/go-ethereum/crypto" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/pkg/jsonrpc2" + "github.com/smartcontractkit/chainlink/v2/core/utils" +) + +const testWorkflowID = "0x1010101010101010101010101010101010101010101010101010101010101010" + +func newTestKey(t *testing.T) *ecdsa.PrivateKey { + t.Helper() + key, err := crypto.GenerateKey() + require.NoError(t, err) + return key +} + +// decodeAsGateway mirrors how the Gateway decodes the wire body, so assertions run against +// what it would actually verify. +func decodeAsGateway(t *testing.T, signed *signedRequest) jsonrpc2.Request[json.RawMessage] { + t.Helper() + req, err := jsonrpc2.DecodeRequest[json.RawMessage](signed.Body, signed.Token) + require.NoError(t, err) + return req +} + +func decodeSegment(t *testing.T, segment string, into any) { + t.Helper() + raw, err := base64.RawURLEncoding.DecodeString(segment) + require.NoError(t, err) + require.NoError(t, json.Unmarshal(raw, into)) +} + +func TestNewSignedRequest_JWTRecoversToSignerAndDigestMatchesBody(t *testing.T) { + key := newTestKey(t) + signed, err := newSignedRequest(key, testWorkflowID, json.RawMessage(`{"orderId":42}`)) + require.NoError(t, err) + + req := decodeAsGateway(t, signed) + claims, recovered, err := utils.VerifyRequestJWT(signed.Token, req) + require.NoError(t, err) + + assert.Equal(t, crypto.PubkeyToAddress(key.PublicKey), recovered) + assert.Equal(t, signed.Signer, recovered) + assert.Equal(t, recovered.Hex(), claims.Issuer) + + digest, err := req.Digest() + require.NoError(t, err) + assert.Equal(t, "0x"+digest, claims.Digest) +} + +func TestNewSignedRequest_JWTHeaderAndClaims(t *testing.T) { + key := newTestKey(t) + before := time.Now().Add(-time.Second) + signed, err := newSignedRequest(key, testWorkflowID, json.RawMessage(`{}`)) + require.NoError(t, err) + + parts := strings.Split(signed.Token, ".") + require.Len(t, parts, 3) + + var header map[string]string + decodeSegment(t, parts[0], &header) + assert.Equal(t, map[string]string{"alg": "ETH", "typ": "JWT"}, header) + + var claims utils.JWTClaims + decodeSegment(t, parts[1], &claims) + _, parseErr := uuid.Parse(claims.ID) + assert.NoError(t, parseErr, "jti must be a UUID") + assert.True(t, claims.IssuedAt.After(before)) + assert.Equal(t, 5*time.Minute, claims.ExpiresAt.Sub(claims.IssuedAt.Time), "lifetime must stay within the Gateway's maximum") +} + +func TestNewSignedRequest_TamperedBodyFailsDigestCheck(t *testing.T) { + key := newTestKey(t) + signed, err := newSignedRequest(key, testWorkflowID, json.RawMessage(`{"orderId":42}`)) + require.NoError(t, err) + + tampered := strings.Replace(string(signed.Body), "42", "43", 1) + require.NotEqual(t, string(signed.Body), tampered) + req, err := jsonrpc2.DecodeRequest[json.RawMessage]([]byte(tampered), signed.Token) + require.NoError(t, err) + + _, _, err = utils.VerifyRequestJWT(signed.Token, req) + require.Error(t, err) + assert.Contains(t, err.Error(), "does not match calculated request digest") +} + +func TestNewSignedRequest_WireShape(t *testing.T) { + signed, err := newSignedRequest(newTestKey(t), testWorkflowID, json.RawMessage(`{"a":[1,2]}`)) + require.NoError(t, err) + + var wire map[string]json.RawMessage + require.NoError(t, json.Unmarshal(signed.Body, &wire)) + assert.JSONEq(t, `"2.0"`, string(wire["jsonrpc"])) + assert.JSONEq(t, `"workflows.execute"`, string(wire["method"])) + assert.JSONEq(t, `"`+signed.ID+`"`, string(wire["id"])) + assert.NotContains(t, wire, "auth", "the JWT travels in the Authorization header, not the body") + assert.JSONEq(t, `{"input":{"a":[1,2]},"workflow":{"workflowID":"`+testWorkflowID+`"}}`, string(wire["params"])) + _, idErr := uuid.Parse(signed.ID) + assert.NoError(t, idErr) + assert.NotContains(t, signed.ID, "/", "the Gateway reserves '/' in request IDs") +} + +func TestNewSignedRequest_KeepsInputNumbersVerbatim(t *testing.T) { + signed, err := newSignedRequest(newTestKey(t), testWorkflowID, json.RawMessage(`{"amount":12345678901234567890}`)) + require.NoError(t, err) + + assert.Contains(t, string(signed.Body), `"amount":12345678901234567890`) +} + +func TestNewSignedRequest_RequestIDAndJTIAreUniquePerCall(t *testing.T) { + key := newTestKey(t) + first, err := newSignedRequest(key, testWorkflowID, json.RawMessage(`{}`)) + require.NoError(t, err) + second, err := newSignedRequest(key, testWorkflowID, json.RawMessage(`{}`)) + require.NoError(t, err) + + assert.NotEqual(t, first.ID, second.ID) + + jti := func(token string) string { + var claims utils.JWTClaims + decodeSegment(t, strings.Split(token, ".")[1], &claims) + return claims.ID + } + assert.NotEqual(t, jti(first.Token), jti(second.Token), "a reused jti is rejected as a replay") +} diff --git a/cmd/workflow/trigger/http/resolve.go b/cmd/workflow/trigger/http/resolve.go new file mode 100644 index 00000000..aa507bb9 --- /dev/null +++ b/cmd/workflow/trigger/http/resolve.go @@ -0,0 +1,109 @@ +package http + +import ( + "context" + "fmt" + "strings" + + "github.com/smartcontractkit/cre-cli/internal/client/graphqlclient" + "github.com/smartcontractkit/cre-cli/internal/client/workflowdataclient" + "github.com/smartcontractkit/cre-cli/internal/tenantctx" + "github.com/smartcontractkit/cre-cli/internal/ui" + "github.com/smartcontractkit/cre-cli/internal/workflowresolve" +) + +// resolveWorkflowID looks up the deployed workflow ID for the workflow-name configured under +// the selected --target, scoped the same way `cre workflow get` scopes it. +func (h *Handler) resolveWorkflowID(ctx context.Context) (string, error) { + rt := h.runtimeContext + if rt.TenantContext == nil { + return "", fmt.Errorf("user context not available — run `cre login` and retry") + } + + name := strings.TrimSpace(rt.Settings.Workflow.UserWorkflowSettings.WorkflowName) + if name == "" { + return "", fmt.Errorf("workflow-name is not set for target %q in workflow.yaml", rt.Settings.User.TargetName) + } + + owner, err := workflowresolve.ResolveWorkflowOwnerAddress(rt.Settings, rt.ResolvedRegistry, rt.DerivedWorkflowOwner) + if err != nil { + return "", err + } + + registry, err := h.registryFilter() + if err != nil { + return "", err + } + + wdc := workflowdataclient.New(graphqlclient.New(rt.Credentials, rt.EnvironmentSet, rt.Logger), rt.Logger) + + spinner := ui.NewSpinner() + spinner.Start(fmt.Sprintf("Resolving workflow %q...", name)) + rows, err := wdc.SearchByName(ctx, name, workflowdataclient.DefaultPageSize, owner) + spinner.Stop() + if err != nil { + return "", fmt.Errorf("resolving workflow name %q: %w", name, err) + } + + rows = workflowresolve.OmitDeleted(exactNameMatches(rows, name)) + if registry != nil { + rows = workflowresolve.FilterRowsByRegistry(rows, registry, rt.TenantContext.Registries) + } + + row, err := pickWorkflow(rows, name) + if err != nil { + return "", err + } + return normalizeWorkflowID(row.WorkflowID) +} + +func (h *Handler) registryFilter() (*tenantctx.Registry, error) { + rt := h.runtimeContext + id := strings.TrimSpace(rt.Settings.Workflow.UserWorkflowSettings.DeploymentRegistry) + if id == "" && rt.ResolvedRegistry != nil { + id = rt.ResolvedRegistry.ID() + } + if id == "" { + return nil, nil + } + + registry := workflowresolve.FindRegistry(rt.TenantContext.Registries, id) + if registry == nil { + return nil, fmt.Errorf("deployment-registry %q not found in user context; available: [%s]", + id, workflowresolve.AvailableRegistryIDs(rt.TenantContext.Registries)) + } + return registry, nil +} + +func exactNameMatches(rows []workflowresolve.Workflow, name string) []workflowresolve.Workflow { + out := make([]workflowresolve.Workflow, 0, len(rows)) + for _, r := range rows { + if strings.EqualFold(strings.TrimSpace(r.Name), name) { + out = append(out, r) + } + } + return out +} + +func pickWorkflow(rows []workflowresolve.Workflow, name string) (workflowresolve.Workflow, error) { + if len(rows) == 0 { + return workflowresolve.Workflow{}, fmt.Errorf("no workflow found with name %q; deploy it first or pass --workflow-id", name) + } + + var active []workflowresolve.Workflow + for _, r := range rows { + if strings.EqualFold(r.Status, "ACTIVE") { + active = append(active, r) + } + } + + switch { + case len(active) == 1: + return active[0], nil + case len(active) > 1: + return workflowresolve.Workflow{}, fmt.Errorf("multiple ACTIVE workflows named %q found; pass --workflow-id to pick one", name) + default: + ui.Warning(fmt.Sprintf("No ACTIVE deployment for workflow %q; using the first match (status: %s)", name, rows[0].Status)) + return rows[0], nil + } +} diff --git a/cmd/workflow/trigger/http/response.go b/cmd/workflow/trigger/http/response.go new file mode 100644 index 00000000..8e665d6a --- /dev/null +++ b/cmd/workflow/trigger/http/response.go @@ -0,0 +1,109 @@ +package http + +import ( + "encoding/json" + "fmt" + "io" + + "github.com/smartcontractkit/chainlink-common/pkg/jsonrpc2" + gwtypes "github.com/smartcontractkit/chainlink-common/pkg/types/gateway" + + "github.com/smartcontractkit/cre-cli/internal/ui" +) + +const ( + statusError = "ERROR" + maxErrorBodyLength = 512 +) + +// output is the machine-readable result printed to stdout with --json. +type output struct { + WorkflowID string `json:"workflowId"` + RequestID string `json:"requestId"` + GatewayURL string `json:"gatewayUrl"` + Status string `json:"status"` + Result json.RawMessage `json:"result"` + Error *jsonrpc2.WireError `json:"error,omitempty"` +} + +// parseGatewayResponse maps the raw POST outcome to a JSON-RPC result or error. The Gateway +// returns its JSON-RPC errors with non-200 HTTP statuses, so the body is inspected first. +func parseGatewayResponse(body []byte, postErr error) (json.RawMessage, *jsonrpc2.WireError) { + var resp jsonrpc2.Response[json.RawMessage] + if err := json.Unmarshal(body, &resp); err == nil && resp.Version == jsonrpc2.JsonRpcVersion { + if resp.Error != nil { + return nil, resp.Error + } + if postErr == nil { + if resp.Result == nil { + return nil, nil + } + return *resp.Result, nil + } + } + + if postErr != nil { + return nil, &jsonrpc2.WireError{Code: jsonrpc2.ErrUnknown, Message: truncate(postErr.Error(), maxErrorBodyLength)} + } + return nil, &jsonrpc2.WireError{ + Code: jsonrpc2.ErrUnknown, + Message: "gateway returned a response that is not JSON-RPC: " + truncate(string(body), maxErrorBodyLength), + } +} + +func newOutput(workflowID, requestID, gatewayURL string, result json.RawMessage, rpcErr *jsonrpc2.WireError) output { + out := output{ + WorkflowID: workflowID, + RequestID: requestID, + GatewayURL: gatewayURL, + Status: resultStatus(result), + Result: result, + Error: rpcErr, + } + if rpcErr != nil { + out.Status = statusError + } + return out +} + +// resultStatus prefers the status reported by the workflow node and assumes ACCEPTED otherwise, +// since a JSON-RPC success from the Gateway means the trigger was delivered. +func resultStatus(result json.RawMessage) string { + var parsed struct { + Status string `json:"status"` + } + if err := json.Unmarshal(result, &parsed); err == nil && parsed.Status != "" { + return parsed.Status + } + return string(gwtypes.HTTPTriggerStatusAccepted) +} + +func printJSON(w io.Writer, out output) error { + data, err := json.MarshalIndent(out, "", " ") + if err != nil { + return fmt.Errorf("failed to encode output: %w", err) + } + _, err = fmt.Fprintln(w, string(data)) + return err +} + +func printSummary(out output) { + ui.Line() + ui.Success(fmt.Sprintf("Trigger request sent (status: %s)", out.Status)) + ui.Line() + ui.Bold("Details:") + ui.Dim(fmt.Sprintf(" Workflow ID: %s", out.WorkflowID)) + ui.Dim(fmt.Sprintf(" Request ID: %s", out.RequestID)) + ui.Dim(fmt.Sprintf(" Gateway: %s", out.GatewayURL)) + if len(out.Result) > 0 { + ui.Dim(fmt.Sprintf(" Result: %s", string(out.Result))) + } + ui.Line() +} + +func truncate(s string, limit int) string { + if len(s) <= limit { + return s + } + return s[:limit] + "..." +} diff --git a/cmd/workflow/trigger/http/response_test.go b/cmd/workflow/trigger/http/response_test.go new file mode 100644 index 00000000..710a71ee --- /dev/null +++ b/cmd/workflow/trigger/http/response_test.go @@ -0,0 +1,123 @@ +package http + +import ( + "bytes" + "encoding/json" + "errors" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/pkg/jsonrpc2" +) + +func TestParseGatewayResponse(t *testing.T) { + tests := []struct { + name string + body string + postErr error + wantResult string + wantCode int64 + wantMsg string + }{ + { + name: "success with result", + body: `{"jsonrpc":"2.0","id":"r1","method":"workflows.execute","result":{"status":"ACCEPTED","workflow_execution_id":"0xabc"}}`, + wantResult: `{"status":"ACCEPTED","workflow_execution_id":"0xabc"}`, + }, + { + name: "success without result", + body: `{"jsonrpc":"2.0","id":"r1"}`, + }, + { + name: "JSON-RPC error on HTTP 200", + body: `{"jsonrpc":"2.0","id":"r1","error":{"code":-32600,"message":"Workflow not found"}}`, + wantCode: -32600, + wantMsg: "Workflow not found", + }, + { + name: "JSON-RPC error carried by a non-200 status", + body: `{"jsonrpc":"2.0","id":"r1","error":{"code":-32600,"message":"Auth failure"}}`, + postErr: errors.New("gateway POST failed: gateway returned non-200: status_code=400"), + wantCode: -32600, + wantMsg: "Auth failure", + }, + { + name: "transport error", + postErr: errors.New("HTTP request to gateway failed: connection refused"), + wantCode: jsonrpc2.ErrUnknown, + wantMsg: "connection refused", + }, + { + name: "non-JSON-RPC error page", + body: `bad gateway`, + postErr: errors.New("gateway returned non-200: status_code=502, body=bad gateway"), + wantCode: jsonrpc2.ErrUnknown, + wantMsg: "status_code=502", + }, + { + name: "HTTP 200 that is not JSON-RPC", + body: `{"hello":"world"}`, + wantCode: jsonrpc2.ErrUnknown, + wantMsg: "not JSON-RPC", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result, rpcErr := parseGatewayResponse([]byte(tt.body), tt.postErr) + + if tt.wantMsg == "" { + require.Nil(t, rpcErr) + if tt.wantResult == "" { + assert.Nil(t, result) + } else { + assert.JSONEq(t, tt.wantResult, string(result)) + } + return + } + require.NotNil(t, rpcErr) + assert.Nil(t, result) + assert.Equal(t, tt.wantCode, rpcErr.Code) + assert.Contains(t, rpcErr.Message, tt.wantMsg) + }) + } +} + +func TestParseGatewayResponse_TruncatesHugeBodies(t *testing.T) { + _, rpcErr := parseGatewayResponse([]byte(strings.Repeat("x", 10_000)), nil) + + require.NotNil(t, rpcErr) + assert.Less(t, len(rpcErr.Message), 1_000) +} + +func TestNewOutput_Status(t *testing.T) { + t.Run("defaults to ACCEPTED", func(t *testing.T) { + assert.Equal(t, "ACCEPTED", newOutput("0x1", "r", "u", json.RawMessage(`{}`), nil).Status) + assert.Equal(t, "ACCEPTED", newOutput("0x1", "r", "u", nil, nil).Status) + }) + t.Run("uses the node-reported status", func(t *testing.T) { + assert.Equal(t, "QUEUED", newOutput("0x1", "r", "u", json.RawMessage(`{"status":"QUEUED"}`), nil).Status) + }) + t.Run("ERROR on JSON-RPC error", func(t *testing.T) { + out := newOutput("0x1", "r", "u", nil, &jsonrpc2.WireError{Code: -32603, Message: "boom"}) + assert.Equal(t, "ERROR", out.Status) + }) +} + +func TestPrintJSON_Shape(t *testing.T) { + var buf bytes.Buffer + require.NoError(t, printJSON(&buf, newOutput("0xabc", "req-1", "https://gw.example", nil, nil))) + + var got map[string]any + require.NoError(t, json.Unmarshal(buf.Bytes(), &got)) + assert.Equal(t, map[string]any{ + "workflowId": "0xabc", + "requestId": "req-1", + "gatewayUrl": "https://gw.example", + "status": "ACCEPTED", + "result": nil, + }, got) +} diff --git a/cmd/workflow/trigger/trigger.go b/cmd/workflow/trigger/trigger.go new file mode 100644 index 00000000..a6cdca83 --- /dev/null +++ b/cmd/workflow/trigger/trigger.go @@ -0,0 +1,22 @@ +package trigger + +import ( + "github.com/spf13/cobra" + + httptrigger "github.com/smartcontractkit/cre-cli/cmd/workflow/trigger/http" + "github.com/smartcontractkit/cre-cli/internal/runtime" +) + +// New returns the hidden `trigger` command group wired under `cre workflow`. +func New(runtimeContext *runtime.Context) *cobra.Command { + triggerCmd := &cobra.Command{ + Use: "trigger", + Short: "Fires the triggers of deployed workflows", + Long: `The trigger command fires a deployed workflow's trigger through the CRE Gateway.`, + Hidden: true, // Internal tooling for live E2E suites; not part of the public CLI surface. + } + + triggerCmd.AddCommand(httptrigger.New(runtimeContext)) + + return triggerCmd +} diff --git a/cmd/workflow/trigger/trigger_test.go b/cmd/workflow/trigger/trigger_test.go new file mode 100644 index 00000000..9ce6823b --- /dev/null +++ b/cmd/workflow/trigger/trigger_test.go @@ -0,0 +1,52 @@ +package trigger_test + +import ( + "bytes" + "testing" + + "github.com/spf13/cobra" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/cre-cli/cmd/workflow/trigger" + "github.com/smartcontractkit/cre-cli/internal/runtime" +) + +func TestNew_CommandsAreHiddenButRunnable(t *testing.T) { + triggerCmd := trigger.New(&runtime.Context{}) + + assert.True(t, triggerCmd.Hidden, "trigger group must not be advertised") + + httpCmd, _, err := triggerCmd.Find([]string{"http"}) + require.NoError(t, err) + require.Equal(t, "http", httpCmd.Name()) + assert.True(t, httpCmd.Hidden, "http subcommand must not be advertised") + assert.NotNil(t, httpCmd.RunE, "hidden must not mean disabled") + assert.Error(t, httpCmd.Args(httpCmd, nil), "the workflow folder argument is required") + assert.NoError(t, httpCmd.Args(httpCmd, []string{"./my-workflow"})) +} + +func TestNew_ParentHelpDoesNotListTrigger(t *testing.T) { + workflowCmd := &cobra.Command{Use: "workflow", Short: "Manages workflows"} + workflowCmd.AddCommand(&cobra.Command{Use: "deploy", Short: "Deploys a workflow", Run: func(*cobra.Command, []string) {}}) + workflowCmd.AddCommand(trigger.New(&runtime.Context{})) + + var help bytes.Buffer + workflowCmd.SetOut(&help) + require.NoError(t, workflowCmd.Help()) + + assert.Contains(t, help.String(), "deploy") + assert.NotContains(t, help.String(), "trigger") +} + +func TestNew_HTTPFlags(t *testing.T) { + httpCmd, _, err := trigger.New(&runtime.Context{}).Find([]string{"http"}) + require.NoError(t, err) + + input := httpCmd.Flags().Lookup("input") + require.NotNil(t, input) + assert.Equal(t, "{}", input.DefValue) + for _, name := range []string{"workflow-id", "output", "json"} { + assert.NotNil(t, httpCmd.Flags().Lookup(name), "missing --%s", name) + } +} diff --git a/cmd/workflow/workflow.go b/cmd/workflow/workflow.go index c301b83a..3352ec41 100644 --- a/cmd/workflow/workflow.go +++ b/cmd/workflow/workflow.go @@ -16,6 +16,7 @@ import ( "github.com/smartcontractkit/cre-cli/cmd/workflow/simulate" supported_chains "github.com/smartcontractkit/cre-cli/cmd/workflow/supported_chains" "github.com/smartcontractkit/cre-cli/cmd/workflow/test" + "github.com/smartcontractkit/cre-cli/cmd/workflow/trigger" "github.com/smartcontractkit/cre-cli/internal/runtime" ) @@ -39,6 +40,7 @@ func New(runtimeContext *runtime.Context) *cobra.Command { workflowCmd.AddCommand(limits.New()) workflowCmd.AddCommand(workflowlist.New(runtimeContext)) workflowCmd.AddCommand(workflowget.New(runtimeContext)) + workflowCmd.AddCommand(trigger.New(runtimeContext)) return workflowCmd } diff --git a/internal/redact/redact.go b/internal/redact/redact.go index ee2d749f..a80159d4 100644 --- a/internal/redact/redact.go +++ b/internal/redact/redact.go @@ -19,6 +19,7 @@ var ( "env": {}, "public-env": {}, "http-payload": {}, + "input": {}, "public_key": {}, "ledger-derivation-path": {}, "config": {}, diff --git a/internal/redact/redact_test.go b/internal/redact/redact_test.go index a8546626..f011ac68 100644 --- a/internal/redact/redact_test.go +++ b/internal/redact/redact_test.go @@ -67,6 +67,7 @@ func TestFlag(t *testing.T) { {name: "env path", flagName: "env", value: "/home/user/project/.env", want: RedactedValue}, {name: "public env path", flagName: "public-env", value: "/tmp/.env.public", want: RedactedValue}, {name: "http payload", flagName: "http-payload", value: `{"secret":"value"}`, want: RedactedValue}, + {name: "trigger input file", flagName: "input", value: "@/home/user/payload.json", want: RedactedValue}, {name: "public key flag", flagName: "public_key", value: "0xabc", want: RedactedValue}, {name: "config path", flagName: "config", value: "./config.yaml", want: RedactedValue}, {name: "limits default", flagName: "limits", value: "default", want: "default"},