Skip to content

Commit baaa22f

Browse files
forestileaoforestileao
authored
fix: validate canvas edge output channels on API writes (#4362)
Closes: #4318 ## Summary Reject canvas edges that reference an output channel the source node does not expose. This prevents silent workflow stalls caused by invalid channel names in CLI/API canvas definitions, and returns a `400 InvalidArgument` instead. ## Changes - Add shared canvas output-channel validation for source nodes - Validate edge channels during full canvas parsing on create/update - Validate edge channels during canvas changeset application - Support validation for: - components - triggers (`default`) - blueprints - Add a transaction-safe blueprint lookup helper for validation paths - Add tests covering invalid channel rejection in: - canvas create - canvas version update - canvas patcher changesets ## Behavior Before: - Invalid edge channels could be persisted - Downstream nodes would silently never execute After: - The API rejects invalid edge channels with `400` - Error message includes the source node and requested channel ## Testing - Ran `make format.go` - Ran `make lint` - Ran `make check.build.app` - Ran `go test -run '^$' ./pkg/grpc/actions/canvases/... ./pkg/models/...` Note: DB-backed tests in this environment could not run because the test database connection was not configured (`lookup port=: no such host`). --------- Signed-off-by: Pedro F. Leao <pedroforestileao@gmail.com>
1 parent df92090 commit baaa22f

8 files changed

Lines changed: 385 additions & 5 deletions

File tree

‎pkg/grpc/actions/canvases/changesets/canvas_patcher.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -408,6 +408,14 @@ func (p *CanvasPatcher) addEdge(change *pb.CanvasChangeset_Change) error {
408408
return fmt.Errorf("target node %s not found", edge.GetTargetId())
409409
}
410410

411+
if err := ValidateSourceNodeOutputChannel(
412+
p.registry,
413+
p.nodes[edge.GetSourceId()],
414+
edge.GetChannel(),
415+
); err != nil {
416+
return err
417+
}
418+
411419
edgeKey := p.edgeKey(edge.GetSourceId(), edge.GetTargetId(), edge.GetChannel())
412420
if _, exists := p.edges[edgeKey]; exists {
413421
return nil

‎pkg/grpc/actions/canvases/changesets/canvas_patcher_test.go‎

Lines changed: 51 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ func Test__CanvasPatcher(t *testing.T) {
4242
},
4343
},
4444
},
45-
[]models.Edge{{SourceID: "node-a", TargetID: "node-b", Channel: "default"}},
45+
[]models.Edge{{SourceID: "node-a", TargetID: "node-b", Channel: "true"}},
4646
)
4747

4848
steps.whenHandling(&pb.CanvasChangeset{
@@ -69,7 +69,7 @@ func Test__CanvasPatcher(t *testing.T) {
6969
Edge: &pb.CanvasChangeset_Change_Edge{
7070
SourceId: "node-a",
7171
TargetId: "node-c",
72-
Channel: "default",
72+
Channel: "true",
7373
},
7474
},
7575
{
@@ -85,7 +85,7 @@ func Test__CanvasPatcher(t *testing.T) {
8585
steps.assertHasNodeBlock("node-c", "noop")
8686
steps.assertHasNoNodeIntegrationID("node-c")
8787
steps.assertNodeCount(2)
88-
steps.assertHasEdge("node-a", "node-c", "default")
88+
steps.assertHasEdge("node-a", "node-c", "true")
8989
steps.assertEdgeCount(1)
9090
steps.assertGraphIsValid()
9191
})
@@ -205,6 +205,54 @@ func Test__CanvasPatcher(t *testing.T) {
205205
steps.assertNodePosition("node-b", 780, 95)
206206
})
207207

208+
t.Run("rejects edges that reference an undefined source output channel", func(t *testing.T) {
209+
steps := &CanvasPatcherSteps{t: t, registry: r.Registry, orgID: r.Organization.ID}
210+
steps.givenCanvasVersion(
211+
[]models.Node{
212+
{
213+
ID: "http-1",
214+
Name: "HTTP Request",
215+
Type: models.NodeTypeComponent,
216+
Ref: models.NodeRef{
217+
Component: &models.ComponentRef{Name: "http"},
218+
},
219+
Configuration: map[string]any{
220+
"method": "GET",
221+
"url": "https://example.com",
222+
},
223+
},
224+
{
225+
ID: "if-1",
226+
Name: "If",
227+
Type: models.NodeTypeComponent,
228+
Ref: models.NodeRef{
229+
Component: &models.ComponentRef{Name: "if"},
230+
},
231+
Configuration: map[string]any{
232+
"expression": "true",
233+
},
234+
},
235+
},
236+
nil,
237+
)
238+
239+
steps.whenHandling(&pb.CanvasChangeset{
240+
Changes: []*pb.CanvasChangeset_Change{
241+
{
242+
Type: pb.CanvasChangeset_Change_ADD_EDGE,
243+
Edge: &pb.CanvasChangeset_Change_Edge{
244+
SourceId: "http-1",
245+
TargetId: "if-1",
246+
Channel: "default",
247+
},
248+
},
249+
},
250+
}, nil)
251+
252+
steps.assertHasError()
253+
steps.assertErrorContains(`source node http-1 does not have output channel "default"`)
254+
})
255+
208256
t.Run("returns error when change object is misconfigured", func(t *testing.T) {
209257
testCases := []struct {
210258
name string

‎pkg/grpc/actions/canvases/changesets/output_channels.go‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
package changesets
2+
3+
import (
4+
"errors"
5+
"fmt"
6+
"slices"
7+
8+
"github.com/superplanehq/superplane/pkg/core"
9+
"github.com/superplanehq/superplane/pkg/models"
10+
"github.com/superplanehq/superplane/pkg/registry"
11+
)
12+
13+
var errUnresolvableSourceNodeOutputChannels = errors.New("source node output channels are not resolvable")
14+
15+
func ValidateSourceNodeOutputChannel(
16+
registry *registry.Registry,
17+
sourceNode models.Node,
18+
channel string,
19+
) error {
20+
outputChannels, err := listSourceNodeOutputChannels(registry, sourceNode)
21+
if err != nil {
22+
if errors.Is(err, errUnresolvableSourceNodeOutputChannels) {
23+
return nil
24+
}
25+
26+
return fmt.Errorf("failed to resolve output channels for source node %s: %w", sourceNode.ID, err)
27+
}
28+
29+
if len(outputChannels) == 0 {
30+
return nil
31+
}
32+
33+
if slices.ContainsFunc(outputChannels, func(outputChannel core.OutputChannel) bool {
34+
return outputChannel.Name == channel
35+
}) {
36+
return nil
37+
}
38+
39+
available := make([]string, 0, len(outputChannels))
40+
for _, outputChannel := range outputChannels {
41+
available = append(available, outputChannel.Name)
42+
}
43+
44+
return fmt.Errorf(
45+
"source node %s does not have output channel %q (available: %v)",
46+
sourceNode.ID,
47+
channel,
48+
available,
49+
)
50+
}
51+
52+
func listSourceNodeOutputChannels(
53+
registry *registry.Registry,
54+
sourceNode models.Node,
55+
) ([]core.OutputChannel, error) {
56+
if sourceNode.Type == models.NodeTypeComponent {
57+
return listComponentOutputChannels(registry, sourceNode)
58+
}
59+
60+
if sourceNode.Type == models.NodeTypeTrigger {
61+
return []core.OutputChannel{core.DefaultOutputChannel}, nil
62+
}
63+
64+
if sourceNode.Type == models.NodeTypeBlueprint {
65+
// TODO: Validate blueprint output channels without doing a blueprint lookup per edge.
66+
return nil, nil
67+
}
68+
69+
return nil, fmt.Errorf("node type %s is not supported", sourceNode.Type)
70+
}
71+
72+
func listComponentOutputChannels(registry *registry.Registry, sourceNode models.Node) ([]core.OutputChannel, error) {
73+
if sourceNode.Ref.Component == nil || sourceNode.Ref.Component.Name == "" {
74+
return nil, fmt.Errorf("%w: component reference is required", errUnresolvableSourceNodeOutputChannels)
75+
}
76+
77+
component, err := registry.GetComponent(sourceNode.Ref.Component.Name)
78+
if err != nil {
79+
return nil, fmt.Errorf("%w: %v", errUnresolvableSourceNodeOutputChannels, err)
80+
}
81+
82+
outputChannels := component.OutputChannels(sourceNode.Configuration)
83+
if len(outputChannels) > 0 {
84+
return outputChannels, nil
85+
}
86+
87+
return []core.OutputChannel{core.DefaultOutputChannel}, nil
88+
}

‎pkg/grpc/actions/canvases/changesets/output_channels_test.go‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
package changesets
2+
3+
import (
4+
"net/http"
5+
"testing"
6+
7+
"github.com/google/uuid"
8+
"github.com/stretchr/testify/assert"
9+
"github.com/stretchr/testify/require"
10+
"github.com/superplanehq/superplane/pkg/configuration"
11+
"github.com/superplanehq/superplane/pkg/core"
12+
"github.com/superplanehq/superplane/pkg/crypto"
13+
"github.com/superplanehq/superplane/pkg/models"
14+
"github.com/superplanehq/superplane/pkg/registry"
15+
)
16+
17+
type testOutputChannelComponent struct {
18+
channels []core.OutputChannel
19+
}
20+
21+
func (c *testOutputChannelComponent) Name() string { return "test-component" }
22+
23+
func (c *testOutputChannelComponent) Label() string { return "Test Component" }
24+
25+
func (c *testOutputChannelComponent) Description() string { return "" }
26+
27+
func (c *testOutputChannelComponent) Documentation() string { return "" }
28+
29+
func (c *testOutputChannelComponent) Icon() string { return "" }
30+
31+
func (c *testOutputChannelComponent) Color() string { return "" }
32+
33+
func (c *testOutputChannelComponent) ExampleOutput() map[string]any { return nil }
34+
35+
func (c *testOutputChannelComponent) OutputChannels(any) []core.OutputChannel { return c.channels }
36+
37+
func (c *testOutputChannelComponent) Configuration() []configuration.Field { return nil }
38+
39+
func (c *testOutputChannelComponent) Setup(core.SetupContext) error { return nil }
40+
41+
func (c *testOutputChannelComponent) ProcessQueueItem(core.ProcessQueueContext) (*uuid.UUID, error) {
42+
return nil, nil
43+
}
44+
45+
func (c *testOutputChannelComponent) Execute(core.ExecutionContext) error { return nil }
46+
47+
func (c *testOutputChannelComponent) Actions() []core.Action { return nil }
48+
49+
func (c *testOutputChannelComponent) HandleAction(core.ActionContext) error { return nil }
50+
51+
func (c *testOutputChannelComponent) HandleWebhook(core.WebhookRequestContext) (int, *core.WebhookResponseBody, error) {
52+
return http.StatusOK, nil, nil
53+
}
54+
55+
func (c *testOutputChannelComponent) Cancel(core.ExecutionContext) error { return nil }
56+
57+
func (c *testOutputChannelComponent) Cleanup(core.SetupContext) error { return nil }
58+
59+
func TestValidateSourceNodeOutputChannel(t *testing.T) {
60+
reg, err := registry.NewRegistry(&crypto.NoOpEncryptor{}, registry.HTTPOptions{})
61+
require.NoError(t, err)
62+
63+
t.Run("unresolvable source component stays soft", func(t *testing.T) {
64+
err := ValidateSourceNodeOutputChannel(
65+
reg,
66+
models.Node{
67+
ID: "node-a",
68+
Type: models.NodeTypeComponent,
69+
Ref: models.NodeRef{
70+
Component: &models.ComponentRef{Name: "missing-component"},
71+
},
72+
},
73+
"default",
74+
)
75+
76+
require.NoError(t, err)
77+
})
78+
79+
t.Run("wrong channel on resolvable component returns error", func(t *testing.T) {
80+
reg.Components["test-component"] = &testOutputChannelComponent{
81+
channels: []core.OutputChannel{
82+
{Name: "success"},
83+
{Name: "failure"},
84+
},
85+
}
86+
87+
err := ValidateSourceNodeOutputChannel(
88+
reg,
89+
models.Node{
90+
ID: "node-a",
91+
Type: models.NodeTypeComponent,
92+
Ref: models.NodeRef{
93+
Component: &models.ComponentRef{Name: "test-component"},
94+
},
95+
},
96+
"default",
97+
)
98+
99+
require.Error(t, err)
100+
assert.Contains(t, err.Error(), `source node node-a does not have output channel "default"`)
101+
assert.Contains(t, err.Error(), "success")
102+
assert.Contains(t, err.Error(), "failure")
103+
})
104+
}

‎pkg/grpc/actions/canvases/create_canvas_test.go‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,57 @@ func TestCreateCanvasOnFreshOrganization(t *testing.T) {
161161
require.Equal(t, r.Organization.ID, persisted.OrganizationID)
162162
}
163163

164+
func TestCreateCanvasRejectsInvalidEdgeChannel(t *testing.T) {
165+
r := support.Setup(t)
166+
ctx := authentication.SetUserIdInMetadata(context.Background(), r.User.String())
167+
168+
canvas := &pb.Canvas{
169+
Metadata: &pb.Canvas_Metadata{
170+
Name: "Invalid HTTP Channel",
171+
},
172+
Spec: &pb.Canvas_Spec{
173+
Nodes: []*componentpb.Node{
174+
{
175+
Id: "http-1",
176+
Name: "HTTP Request",
177+
Type: componentpb.Node_TYPE_COMPONENT,
178+
Component: &componentpb.Node_ComponentRef{
179+
Name: "http",
180+
},
181+
Configuration: structFromAnyMap(t, map[string]any{
182+
"method": "GET",
183+
"url": "https://example.com",
184+
}),
185+
},
186+
{
187+
Id: "if-1",
188+
Name: "If",
189+
Type: componentpb.Node_TYPE_COMPONENT,
190+
Component: &componentpb.Node_ComponentRef{
191+
Name: "if",
192+
},
193+
Configuration: structFromAnyMap(t, map[string]any{
194+
"expression": "true",
195+
}),
196+
},
197+
},
198+
Edges: []*componentpb.Edge{
199+
{
200+
SourceId: "http-1",
201+
TargetId: "if-1",
202+
Channel: "default",
203+
},
204+
},
205+
},
206+
}
207+
208+
baseURL := "https://example.com"
209+
_, err := CreateCanvas(ctx, r.Registry, r.Encryptor, r.AuthService, baseURL, r.Organization.ID, canvas, nil, nil)
210+
require.Error(t, err)
211+
require.Equal(t, codes.InvalidArgument, status.Code(err))
212+
require.Contains(t, status.Convert(err).Message(), `source node http-1 does not have output channel "default"`)
213+
}
214+
164215
func TestCreateCanvasWithUsageRejectsLimitViolation(t *testing.T) {
165216
r := support.Setup(t)
166217
ctx := authentication.SetUserIdInMetadata(context.Background(), r.User.String())

‎pkg/grpc/actions/canvases/serialization.go‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -262,6 +262,11 @@ func ParseCanvas(registry *registry.Registry, orgID string, canvas *pb.Canvas) (
262262

263263
// Find shadowed names within connected components
264264
nodeWarnings := actions.FindShadowedNameWarnings(canvas.Spec.Nodes, canvas.Spec.Edges)
265+
nodes := actions.ProtoToNodes(canvas.Spec.Nodes)
266+
nodesByID := make(map[string]models.Node, len(nodes))
267+
for _, node := range nodes {
268+
nodesByID[node.ID] = node
269+
}
265270

266271
for i, edge := range canvas.Spec.Edges {
267272
if edge.SourceId == "" || edge.TargetId == "" {
@@ -283,10 +288,17 @@ func ParseCanvas(registry *registry.Registry, orgID string, canvas *pb.Canvas) (
283288
if nodeTypeByID[edge.TargetId] == compb.Node_TYPE_WIDGET {
284289
return nil, nil, status.Errorf(codes.InvalidArgument, "edge %d: widget nodes cannot be used as target nodes", i)
285290
}
291+
292+
if err := changesets.ValidateSourceNodeOutputChannel(
293+
registry,
294+
nodesByID[edge.SourceId],
295+
edge.Channel,
296+
); err != nil {
297+
return nil, nil, status.Errorf(codes.InvalidArgument, "edge %d: %v", i, err)
298+
}
286299
}
287300

288301
// Convert proto nodes to models, adding validation errors and warnings where applicable
289-
nodes := actions.ProtoToNodes(canvas.Spec.Nodes)
290302
edges := actions.ProtoToEdges(canvas.Spec.Edges)
291303
for i := range nodes {
292304
if errorMsg, hasError := nodeValidationErrors[nodes[i].ID]; hasError {

0 commit comments

Comments
 (0)