From d3319decfefa0faa17c53ab083c4a1bcb16837f7 Mon Sep 17 00:00:00 2001 From: Aly Raffauf Date: Tue, 1 Sep 2026 20:58:07 -0400 Subject: [PATCH] app/pipeline: enforce ownership and bounded selection --- internal/app/pipeline.go | 16 +++++- internal/app/pipeline_cancel.go | 25 ++++++++- internal/app/pipeline_list.go | 5 ++ internal/app/pipeline_logs.go | 5 +- internal/app/pipeline_status.go | 27 ++++++++-- internal/app/pipeline_view.go | 10 +--- internal/app/pipelines_test.go | 79 +++++++++++++++++++++++---- internal/cli/pipeline_logs.go | 9 +++- internal/cli/pipeline_logs_test.go | 87 ++++++++++++++++++++++++++++++ 9 files changed, 237 insertions(+), 26 deletions(-) create mode 100644 internal/cli/pipeline_logs_test.go diff --git a/internal/app/pipeline.go b/internal/app/pipeline.go index db3226d..29d15e9 100644 --- a/internal/app/pipeline.go +++ b/internal/app/pipeline.go @@ -7,7 +7,7 @@ import ( "github.com/alyraffauf/tg/spindle" ) -const maxPipelinePages = 1000 +const maxPipelinePages = 100 func (s *Service) pipelineClient(ctx context.Context, target Target) (pipelineClient, string, error) { spindleHost, repoDID, err := s.pipelineTarget(ctx, target) @@ -21,6 +21,20 @@ func (s *Service) pipelineClient(ctx context.Context, target Target) (pipelineCl return client, repoDID, nil } +func fetchOwnedPipeline(ctx context.Context, client pipelineClient, target Target, repoDID, pipelineID string) (*spindle.Pipeline, error) { + pipeline, err := client.GetPipeline(ctx, pipelineID) + if err != nil { + return nil, err + } + if pipeline == nil { + return nil, fmt.Errorf("pipeline %q returned an empty response", pipelineID) + } + if pipeline.Repo == "" || pipeline.Repo != repoDID { + return nil, fmt.Errorf("pipeline %q does not belong to repository %q", pipelineID, target.String()) + } + return pipeline, nil +} + func (s *Service) pipelineTarget(ctx context.Context, target Target) (string, string, error) { repo, err := s.resolveRepo(ctx, target) if err != nil { diff --git a/internal/app/pipeline_cancel.go b/internal/app/pipeline_cancel.go index da3317b..43ba2ba 100644 --- a/internal/app/pipeline_cancel.go +++ b/internal/app/pipeline_cancel.go @@ -17,10 +17,13 @@ func (s *Service) CancelPipeline(ctx context.Context, target Target, pipelineID if err != nil { return nil, fmt.Errorf("connect to pipeline spindle: %w", err) } - pipeline, err := client.GetPipeline(ctx, pipelineID) + pipeline, err := fetchOwnedPipeline(ctx, client, target, repoDID, pipelineID) if err != nil { return nil, err } + if err := validateSelectedWorkflows(pipeline.Workflows, workflows); err != nil { + return nil, err + } cancellableWorkflows := selectCancellableWorkflows(pipeline.Workflows, workflows) if len(cancellableWorkflows) == 0 { return &PipelineCancelResult{Pipeline: pipelineID}, nil @@ -54,6 +57,26 @@ func (s *Service) CancelPipeline(ctx context.Context, target Target, pipelineID return &PipelineCancelResult{Pipeline: pipelineID, Workflows: cancellableWorkflows, CancellationRequested: true}, nil } +func validateSelectedWorkflows(workflows []spindle.Workflow, selected []string) error { + if len(selected) == 0 { + return nil + } + byName := make(map[string]string, len(workflows)) + for _, workflow := range workflows { + byName[workflow.Name] = workflow.Status + } + for _, name := range selected { + status, found := byName[name] + if !found { + return fmt.Errorf("workflow %q was not found in the pipeline", name) + } + if status != "pending" && status != "running" { + return fmt.Errorf("workflow %q is already finished with status %q", name, status) + } + } + return nil +} + func selectCancellableWorkflows(workflows []spindle.Workflow, selected []string) []string { selectedWorkflows := make(map[string]bool, len(selected)) for _, workflow := range selected { diff --git a/internal/app/pipeline_list.go b/internal/app/pipeline_list.go index 0f590f1..e365b75 100644 --- a/internal/app/pipeline_list.go +++ b/internal/app/pipeline_list.go @@ -17,6 +17,7 @@ func (s *Service) ListPipelines(ctx context.Context, target Target) ([]Pipeline, func listPipelinePages(ctx context.Context, client pipelineClient, repoDID string) ([]Pipeline, error) { var pipelines []Pipeline cursor := "" + seenCursors := make(map[string]bool) for page := 0; page < maxPipelinePages; page++ { response, err := client.QueryPipelines(ctx, repoDID, cursor) if err != nil { @@ -26,6 +27,10 @@ func listPipelinePages(ctx context.Context, client pipelineClient, repoDID strin if response.Cursor == "" { return pipelines, nil } + if seenCursors[response.Cursor] { + return nil, fmt.Errorf("pipeline pagination repeated cursor %q", response.Cursor) + } + seenCursors[response.Cursor] = true cursor = response.Cursor } return nil, fmt.Errorf("exceeded %d pipeline pages without reaching the end of the list", maxPipelinePages) diff --git a/internal/app/pipeline_logs.go b/internal/app/pipeline_logs.go index d992f8d..6595abb 100644 --- a/internal/app/pipeline_logs.go +++ b/internal/app/pipeline_logs.go @@ -10,7 +10,7 @@ import ( // PipelineLogs streams log events from a pipeline. A non-empty workflows list // filters to the named workflows. func (s *Service) PipelineLogs(ctx context.Context, target Target, pipelineID string, workflows []string, onEvent func(PipelineLogEvent) error) error { - spindleHost, _, err := s.pipelineTarget(ctx, target) + spindleHost, repoDID, err := s.pipelineTarget(ctx, target) if err != nil { return err } @@ -18,6 +18,9 @@ func (s *Service) PipelineLogs(ctx context.Context, target Target, pipelineID st if err != nil { return fmt.Errorf("connect to pipeline spindle: %w", err) } + if _, err := fetchOwnedPipeline(ctx, client, target, repoDID, pipelineID); err != nil { + return err + } return client.SubscribePipelineLogs(ctx, pipelineID, workflows, func(event spindle.PipelineLogEvent) error { return onEvent(PipelineLogEvent{ Type: event.Type, diff --git a/internal/app/pipeline_status.go b/internal/app/pipeline_status.go index d7f1feb..0950f13 100644 --- a/internal/app/pipeline_status.go +++ b/internal/app/pipeline_status.go @@ -32,18 +32,39 @@ func (s *Service) PipelineStatus(ctx context.Context, target Target) (*PipelineS if err != nil { return nil, fmt.Errorf("connect to pipeline spindle: %w", err) } - response, err := client.QueryPipelines(ctx, repoDID, "") + pipeline, err := findDefaultBranchPipeline(ctx, client, repoDID, defaultBranch.Name, defaultBranch.Hash) if err != nil { return nil, err } - pipelines := pipelineItems(response.Pipelines) - pipeline := latestDefaultBranchPipeline(pipelines, defaultBranch.Name, defaultBranch.Hash) if pipeline == nil { return nil, fmt.Errorf("no pipeline found for the latest %s commit on the default branch", target.String()) } return &PipelineStatusResult{Commit: pipeline.Commit, Pipeline: pipeline, HasFailures: pipelineHasFailures(*pipeline)}, nil } +func findDefaultBranchPipeline(ctx context.Context, client pipelineClient, repoDID, branchName, branchHash string) (*Pipeline, error) { + cursor := "" + seenCursors := make(map[string]bool) + for page := 0; page < maxPipelinePages; page++ { + response, err := client.QueryPipelines(ctx, repoDID, cursor) + if err != nil { + return nil, err + } + if pipeline := latestDefaultBranchPipeline(pipelineItems(response.Pipelines), branchName, branchHash); pipeline != nil { + return pipeline, nil + } + if response.Cursor == "" { + return nil, nil + } + if seenCursors[response.Cursor] { + return nil, fmt.Errorf("pipeline pagination repeated cursor %q", response.Cursor) + } + seenCursors[response.Cursor] = true + cursor = response.Cursor + } + return nil, fmt.Errorf("exceeded %d pipeline pages while searching for the latest default-branch commit", maxPipelinePages) +} + func latestDefaultBranchPipeline(pipelines []Pipeline, branchName, branchHash string) *Pipeline { for index := range pipelines { pipeline := &pipelines[index] diff --git a/internal/app/pipeline_view.go b/internal/app/pipeline_view.go index 72de15b..24ee3ea 100644 --- a/internal/app/pipeline_view.go +++ b/internal/app/pipeline_view.go @@ -1,9 +1,6 @@ package app -import ( - "context" - "fmt" -) +import "context" // ViewPipeline fetches a pipeline by its spindle-local ID. func (s *Service) ViewPipeline(ctx context.Context, target Target, pipelineID string) (*Pipeline, error) { @@ -11,13 +8,10 @@ func (s *Service) ViewPipeline(ctx context.Context, target Target, pipelineID st if err != nil { return nil, err } - pipeline, err := client.GetPipeline(ctx, pipelineID) + pipeline, err := fetchOwnedPipeline(ctx, client, target, repoDID, pipelineID) if err != nil { return nil, err } - if pipeline.Repo != repoDID { - return nil, fmt.Errorf("pipeline %q does not belong to repository %q", pipelineID, target.String()) - } item := pipelineItem(*pipeline) return &item, nil } diff --git a/internal/app/pipelines_test.go b/internal/app/pipelines_test.go index 7437ba7..365fc4b 100644 --- a/internal/app/pipelines_test.go +++ b/internal/app/pipelines_test.go @@ -127,8 +127,61 @@ func TestPipelineStatusReturnsLatestPipeline(t *testing.T) { } } +func TestPipelineStatusFindsCommitOnLaterPage(t *testing.T) { + client := &testPipelineClient{responses: []*spindle.QueryPipelinesOutput{ + {Pipelines: []spindle.Pipeline{{ID: "older", Commit: "old"}}, Cursor: "next"}, + {Pipelines: []spindle.Pipeline{{ID: "latest", Commit: "abc"}}}, + }} + service := testService(&testPDS{}, &testGit{}, &testKnot{defaultBranch: &knot.DefaultBranch{Name: "main", Hash: "abc"}}) + service.appview = testAppview{repo: &tangled.Repo{Value: tangledlex.Repo{ + Knot: "knot.example", Spindle: optionalString("spindle.example"), RepoDid: optionalString("did:plc:repo"), + }}} + service.spindle = testSpindleFactory{client: client} + + status, err := service.PipelineStatus(context.Background(), Target{Handle: "owner.test", Repo: "example"}) + if err != nil { + t.Fatalf("PipelineStatus() error = %v", err) + } + if status.Pipeline.ID != "latest" || !slices.Equal(client.cursors, []string{"", "next"}) { + t.Fatalf("status=%+v cursors=%q", status, client.cursors) + } +} + +func TestListPipelinePagesRejectsRepeatedCursor(t *testing.T) { + client := &testPipelineClient{responses: []*spindle.QueryPipelinesOutput{{Cursor: "same"}, {Cursor: "same"}}} + _, err := listPipelinePages(context.Background(), client, "did:plc:repo") + if err == nil || err.Error() != "pipeline pagination repeated cursor \"same\"" { + t.Fatalf("listPipelinePages() error = %v", err) + } +} + +func TestCancelPipelineRejectsExplicitFinishedWorkflowBeforeAuth(t *testing.T) { + client := &testPipelineClient{pipeline: &spindle.Pipeline{Repo: "did:plc:repo", Workflows: []spindle.Workflow{{Name: "done.yml", Status: "success"}}}} + pds := &testPDS{} + service := testService(pds, &testGit{}, &testKnot{}) + service.appview = testAppview{repo: &tangled.Repo{Value: tangledlex.Repo{Spindle: optionalString("spindle.example"), RepoDid: optionalString("did:plc:repo")}}} + service.spindle = testSpindleFactory{client: client} + + _, err := service.CancelPipeline(context.Background(), Target{Handle: "owner.test", Repo: "example"}, "pipeline", []string{"done.yml"}) + if err == nil || pds.serviceAuthCalls != 0 || client.cancelCalls != 0 { + t.Fatalf("CancelPipeline() error=%v auth=%d cancel=%d", err, pds.serviceAuthCalls, client.cancelCalls) + } +} + +func TestPipelineLogsRejectsForeignPipelineBeforeSubscription(t *testing.T) { + client := &testPipelineClient{pipeline: &spindle.Pipeline{Repo: "did:plc:foreign"}} + service := testService(&testPDS{}, &testGit{}, &testKnot{}) + service.appview = testAppview{repo: &tangled.Repo{Value: tangledlex.Repo{Spindle: optionalString("spindle.example"), RepoDid: optionalString("did:plc:repo")}}} + service.spindle = testSpindleFactory{client: client} + + err := service.PipelineLogs(context.Background(), Target{Handle: "owner.test", Repo: "example"}, "pipeline", nil, func(PipelineLogEvent) error { return nil }) + if err == nil || client.subscribeCalls != 0 { + t.Fatalf("PipelineLogs() error=%v subscribe=%d", err, client.subscribeCalls) + } +} + func TestCancelPipelineMintsSpindleToken(t *testing.T) { - client := &testPipelineClient{pipeline: &spindle.Pipeline{Workflows: []spindle.Workflow{ + client := &testPipelineClient{pipeline: &spindle.Pipeline{Repo: "did:plc:repo", Workflows: []spindle.Workflow{ {Name: "test.yml", Status: "pending"}, {Name: "done.yml", Status: "success"}, }}} @@ -139,7 +192,7 @@ func TestCancelPipelineMintsSpindleToken(t *testing.T) { }}} service.spindle = testSpindleFactory{client: client} - result, err := service.CancelPipeline(context.Background(), Target{Handle: "owner.test", Repo: "example"}, "3mrvk5dbnep22", []string{"test.yml", "done.yml", "unknown.yml"}) + result, err := service.CancelPipeline(context.Background(), Target{Handle: "owner.test", Repo: "example"}, "3mrvk5dbnep22", []string{"test.yml"}) if err != nil { t.Fatalf("CancelPipeline() error = %v", err) } @@ -180,15 +233,17 @@ func TestTriggerPipelineUsesFullSHAWithoutGitResolution(t *testing.T) { } type testPipelineClient struct { - responses []*spindle.QueryPipelinesOutput - cursors []string - err error - cancelInput spindle.CancelPipelineInput - pipeline *spindle.Pipeline - pipelineID string - triggerInput spindle.TriggerPipelineInput - triggerOutput *spindle.TriggerPipelineOutput - logEvents []spindle.PipelineLogEvent + responses []*spindle.QueryPipelinesOutput + cursors []string + err error + cancelInput spindle.CancelPipelineInput + pipeline *spindle.Pipeline + pipelineID string + triggerInput spindle.TriggerPipelineInput + triggerOutput *spindle.TriggerPipelineOutput + logEvents []spindle.PipelineLogEvent + cancelCalls int + subscribeCalls int } func (c *testPipelineClient) QueryLatestPipeline(_ context.Context, _ string) (*spindle.QueryPipelinesOutput, error) { @@ -201,6 +256,7 @@ func (c *testPipelineClient) GetPipeline(_ context.Context, pipelineID string) ( } func (c *testPipelineClient) CancelPipeline(_ context.Context, input spindle.CancelPipelineInput) error { + c.cancelCalls++ c.cancelInput = input return c.err } @@ -233,6 +289,7 @@ func (c *testPipelineClient) QueryPipelines(_ context.Context, _ string, cursor } func (c *testPipelineClient) SubscribePipelineLogs(_ context.Context, _ string, _ []string, onEvent func(spindle.PipelineLogEvent) error) error { + c.subscribeCalls++ for _, event := range c.logEvents { if err := onEvent(event); err != nil { return err diff --git a/internal/cli/pipeline_logs.go b/internal/cli/pipeline_logs.go index 988a760..797a81c 100644 --- a/internal/cli/pipeline_logs.go +++ b/internal/cli/pipeline_logs.go @@ -1,6 +1,7 @@ package cli import ( + "context" "encoding/json" "fmt" "image/color" @@ -12,12 +13,18 @@ import ( "github.com/spf13/cobra" ) +type pipelineLogsService interface { + TargetFromCWD(context.Context) (app.Target, error) + PipelineStatus(context.Context, app.Target) (*app.PipelineStatusResult, error) + PipelineLogs(context.Context, app.Target, string, []string, func(app.PipelineLogEvent) error) error +} + var workflowColors = []color.Color{ lipgloss.Cyan, lipgloss.Magenta, lipgloss.Green, lipgloss.Yellow, lipgloss.Blue, lipgloss.Red, } -func newPipelineLogsCommand(service *app.Service) *cobra.Command { +func newPipelineLogsCommand(service pipelineLogsService) *cobra.Command { var repository string var workflows []string diff --git a/internal/cli/pipeline_logs_test.go b/internal/cli/pipeline_logs_test.go new file mode 100644 index 0000000..6ffdd2e --- /dev/null +++ b/internal/cli/pipeline_logs_test.go @@ -0,0 +1,87 @@ +package cli + +import ( + "bytes" + "context" + "errors" + "strings" + "testing" + + "github.com/alyraffauf/tg/internal/app" +) + +type recordingPipelineLogsService struct { + target app.Target + pipeline string + workflows []string + events []app.PipelineLogEvent + err error +} + +func (service *recordingPipelineLogsService) TargetFromCWD(context.Context) (app.Target, error) { + return app.Target{}, errors.New("unexpected target detection") +} + +func (service *recordingPipelineLogsService) PipelineStatus(context.Context, app.Target) (*app.PipelineStatusResult, error) { + return nil, errors.New("unexpected pipeline status lookup") +} + +func (service *recordingPipelineLogsService) PipelineLogs(_ context.Context, target app.Target, pipeline string, workflows []string, callback func(app.PipelineLogEvent) error) error { + service.target = target + service.pipeline = pipeline + service.workflows = append([]string(nil), workflows...) + for _, event := range service.events { + if err := callback(event); err != nil { + return err + } + } + return service.err +} + +func TestPipelineLogsJSONStreamsNDJSONToStdout(t *testing.T) { + service := &recordingPipelineLogsService{events: []app.PipelineLogEvent{ + {Type: "data", Data: &app.PipelineLogData{Stream: "stdout", Content: "out", Workflow: "build"}}, + {Type: "data", Data: &app.PipelineLogData{Stream: "stderr", Content: "err", Workflow: "build"}}, + }} + command := newPipelineLogsCommand(service) + command.Flags().Bool("json", false, "") + command.SetArgs([]string{"pipeline-1", "--repo", "owner.test/example", "--workflow", "build", "--json"}) + var stdout, stderr bytes.Buffer + command.SetOut(&stdout) + command.SetErr(&stderr) + if err := command.Execute(); err != nil { + t.Fatalf("Execute() error = %v", err) + } + lines := strings.Split(strings.TrimSpace(stdout.String()), "\n") + if len(lines) != 2 || !strings.Contains(lines[0], `"content":"out"`) || !strings.Contains(lines[1], `"content":"err"`) { + t.Fatalf("NDJSON output = %q", stdout.String()) + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q", stderr.String()) + } + if service.target.String() != "owner.test/example" || service.pipeline != "pipeline-1" || len(service.workflows) != 1 || service.workflows[0] != "build" { + t.Fatalf("service input: target=%s pipeline=%q workflows=%q", service.target, service.pipeline, service.workflows) + } +} + +func TestPipelineLogsJSONPropagatesStreamAndWriterErrors(t *testing.T) { + streamFailure := errors.New("stream failed") + service := &recordingPipelineLogsService{err: streamFailure} + command := newPipelineLogsCommand(service) + command.Flags().Bool("json", false, "") + command.SetArgs([]string{"pipeline-1", "--repo", "owner.test/example", "--json"}) + command.SetOut(&bytes.Buffer{}) + if err := command.Execute(); !errors.Is(err, streamFailure) { + t.Fatalf("stream error = %v", err) + } + + service.err = nil + service.events = []app.PipelineLogEvent{{Type: "data", Data: &app.PipelineLogData{Content: "line"}}} + command = newPipelineLogsCommand(service) + command.Flags().Bool("json", false, "") + command.SetArgs([]string{"pipeline-1", "--repo", "owner.test/example", "--json"}) + command.SetOut(errorWriter{}) + if err := command.Execute(); !errors.Is(err, errWrite) { + t.Fatalf("writer error = %v", err) + } +} -- 2.51.2