diff --git a/service/internal/api/api.go b/service/internal/api/api.go index 83465aa..3497fb1 100644 --- a/service/internal/api/api.go +++ b/service/internal/api/api.go @@ -60,6 +60,20 @@ type streamingClient struct { AuthenticatedUser *authpublic.AuthenticatedUser } +// trySendEventToClient sends msg to the client's channel. Returns false if the channel is full (client should be removed). +func (api *oliveTinAPI) trySendEventToClient(client *streamingClient, msg *apiv1.EventStreamResponse) bool { + if client == nil || msg == nil { + return false + } + select { + case client.channel <- msg: + return true + default: + log.Warnf("EventStream: client channel is full, removing client") + return false + } +} + func (api *oliveTinAPI) KillAction(ctx ctx.Context, req *connect.Request[apiv1.KillActionRequest]) (*connect.Response[apiv1.KillActionResponse], error) { ret := &apiv1.KillActionResponse{ ExecutionTrackingId: req.Msg.ExecutionTrackingId, @@ -589,9 +603,20 @@ func isValidLogEntry(e *executor.InternalLogEntry) bool { // isLogEntryAllowed checks if a log entry is allowed to be viewed by the user. func (api *oliveTinAPI) isLogEntryAllowed(e *executor.InternalLogEntry, user *authpublic.AuthenticatedUser) bool { + if user == nil || !isValidLogEntry(e) { + return false + } return acl.IsAllowedLogs(api.cfg, user, e.Binding.Action) } +// mayViewExecutionEvent returns whether the user is allowed to receive this execution event (for EventStream ACL). +func (api *oliveTinAPI) mayViewExecutionEvent(entry *executor.InternalLogEntry, user *authpublic.AuthenticatedUser) bool { + if user == nil { + return false + } + return isValidLogEntry(entry) && api.isLogEntryAllowed(entry, user) +} + // buildEmptyPageResponse creates a response for an empty page. func buildEmptyPageResponse(page pageInfo) *apiv1.GetActionLogsResponse { return &apiv1.GetActionLogsResponse{ @@ -886,6 +911,9 @@ func (api *oliveTinAPI) EventStream(ctx ctx.Context, req *connect.Request[apiv1. } func (api *oliveTinAPI) removeClient(clientToRemove *streamingClient) { + if clientToRemove == nil { + return + } api.streamingClientsMutex.Lock() delete(api.streamingClients, clientToRemove) api.streamingClientsMutex.Unlock() @@ -915,50 +943,62 @@ func (api *oliveTinAPI) OnActionMapRebuilt() { func (api *oliveTinAPI) OnExecutionStarted(ex *executor.InternalLogEntry) { toRemove := []*streamingClient{} - for _, client := range api.copyOfStreamingClients() { - select { - case client.channel <- &apiv1.EventStreamResponse{ - Event: &apiv1.EventStreamResponse_ExecutionStarted{ - ExecutionStarted: &apiv1.EventExecutionStarted{ - LogEntry: api.internalLogEntryToPb(ex, client.AuthenticatedUser), - }, - }, - }: - default: - log.Warnf("EventStream: client channel is full, removing client") - toRemove = append(toRemove, client) - } + api.maybeSendExecutionStarted(client, ex, &toRemove) } - for _, client := range toRemove { api.removeClient(client) } } +func (api *oliveTinAPI) maybeSendExecutionStarted(client *streamingClient, ex *executor.InternalLogEntry, toRemove *[]*streamingClient) { + if client == nil { + return + } + if !api.mayViewExecutionEvent(ex, client.AuthenticatedUser) { + return + } + msg := &apiv1.EventStreamResponse{ + Event: &apiv1.EventStreamResponse_ExecutionStarted{ + ExecutionStarted: &apiv1.EventExecutionStarted{ + LogEntry: api.internalLogEntryToPb(ex, client.AuthenticatedUser), + }, + }, + } + if !api.trySendEventToClient(client, msg) { + *toRemove = append(*toRemove, client) + } +} + func (api *oliveTinAPI) OnExecutionFinished(ile *executor.InternalLogEntry) { toRemove := []*streamingClient{} - for _, client := range api.copyOfStreamingClients() { - select { - case client.channel <- &apiv1.EventStreamResponse{ - Event: &apiv1.EventStreamResponse_ExecutionFinished{ - ExecutionFinished: &apiv1.EventExecutionFinished{ - LogEntry: api.internalLogEntryToPb(ile, client.AuthenticatedUser), - }, - }, - }: - default: - log.Warnf("EventStream: client channel is full, removing client") - toRemove = append(toRemove, client) - } + api.maybeSendExecutionFinished(client, ile, &toRemove) } - for _, client := range toRemove { api.removeClient(client) } } +func (api *oliveTinAPI) maybeSendExecutionFinished(client *streamingClient, ile *executor.InternalLogEntry, toRemove *[]*streamingClient) { + if client == nil { + return + } + if !api.mayViewExecutionEvent(ile, client.AuthenticatedUser) { + return + } + msg := &apiv1.EventStreamResponse{ + Event: &apiv1.EventStreamResponse_ExecutionFinished{ + ExecutionFinished: &apiv1.EventExecutionFinished{ + LogEntry: api.internalLogEntryToPb(ile, client.AuthenticatedUser), + }, + }, + } + if !api.trySendEventToClient(client, msg) { + *toRemove = append(*toRemove, client) + } +} + func (api *oliveTinAPI) GetDiagnostics(ctx ctx.Context, req *connect.Request[apiv1.GetDiagnosticsRequest]) (*connect.Response[apiv1.GetDiagnosticsResponse], error) { user := auth.UserFromApiCall(ctx, req, api.cfg) if err := api.checkDashboardAccess(user); err != nil { @@ -1128,29 +1168,47 @@ func buildAdditionalLinks(links []*config.NavigationLink) []*apiv1.AdditionalLin } func (api *oliveTinAPI) OnOutputChunk(content []byte, executionTrackingId string) { - toRemove := []*streamingClient{} - - for _, client := range api.copyOfStreamingClients() { - select { - case client.channel <- &apiv1.EventStreamResponse{ - Event: &apiv1.EventStreamResponse_OutputChunk{ - OutputChunk: &apiv1.EventOutputChunk{ - Output: string(content), - ExecutionTrackingId: executionTrackingId, - }, - }, - }: - default: - log.Warnf("EventStream: client channel is full, removing client") - toRemove = append(toRemove, client) - } + entry := api.getValidLogEntryForStreaming(executionTrackingId) + if entry == nil { + return + } + msg := &apiv1.EventStreamResponse{ + Event: &apiv1.EventStreamResponse_OutputChunk{ + OutputChunk: &apiv1.EventOutputChunk{ + Output: string(content), + ExecutionTrackingId: executionTrackingId, + }, + }, + } + toRemove := []*streamingClient{} + for _, client := range api.copyOfStreamingClients() { + api.maybeSendOutputChunk(client, entry, msg, &toRemove) } - for _, client := range toRemove { api.removeClient(client) } } +func (api *oliveTinAPI) getValidLogEntryForStreaming(executionTrackingId string) *executor.InternalLogEntry { + entry, ok := api.executor.GetLog(executionTrackingId) + if !ok || !isValidLogEntry(entry) { + return nil + } + return entry +} + +func (api *oliveTinAPI) maybeSendOutputChunk(client *streamingClient, entry *executor.InternalLogEntry, msg *apiv1.EventStreamResponse, toRemove *[]*streamingClient) { + if client == nil { + return + } + if !api.mayViewExecutionEvent(entry, client.AuthenticatedUser) { + return + } + if !api.trySendEventToClient(client, msg) { + *toRemove = append(*toRemove, client) + } +} + func (api *oliveTinAPI) GetEntities(ctx ctx.Context, req *connect.Request[apiv1.GetEntitiesRequest]) (*connect.Response[apiv1.GetEntitiesResponse], error) { user := auth.UserFromApiCall(ctx, req, api.cfg) diff --git a/service/internal/api/api_test.go b/service/internal/api/api_test.go index c6ee915..ca8ecf5 100644 --- a/service/internal/api/api_test.go +++ b/service/internal/api/api_test.go @@ -2,24 +2,24 @@ package api import ( "context" + "net/http" + "net/http/httptest" + "path" "testing" + "time" "connectrpc.com/connect" + "github.com/google/uuid" + log "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - log "github.com/sirupsen/logrus" - apiv1 "github.com/OliveTin/OliveTin/gen/olivetin/api/v1" apiv1connect "github.com/OliveTin/OliveTin/gen/olivetin/api/v1/apiv1connect" authpublic "github.com/OliveTin/OliveTin/internal/auth/authpublic" config "github.com/OliveTin/OliveTin/internal/config" "github.com/OliveTin/OliveTin/internal/entities" "github.com/OliveTin/OliveTin/internal/executor" - - "net/http" - "net/http/httptest" - "path" ) func getNewTestServerAndClient(injectedConfig *config.Config) (*httptest.Server, apiv1connect.OliveTinApiServiceClient) { @@ -338,12 +338,13 @@ func testWithEntity(t *testing.T, binding *executor.ActionBinding, rr *Dashboard } // buildViewPermissionTestConfig returns config and users for GHSA view-permission tests: -// one action "secret_action", ACL "restricted" (view:false) for user "low", ACL "full" (view:true) for user "admin". +// one action "secret_action", ACL "restricted" (view:false, logs:false) for user "low", ACL "full" (view:true, logs:true) for user "admin". func buildViewPermissionTestConfig(t *testing.T) (*config.Config, *authpublic.AuthenticatedUser, *authpublic.AuthenticatedUser) { t.Helper() cfg := config.DefaultConfig() cfg.DefaultPermissions.View = false cfg.DefaultPermissions.Exec = false + cfg.DefaultPermissions.Logs = false cfg.Actions = append(cfg.Actions, &config.Action{ ID: "secret_action", @@ -572,3 +573,127 @@ func TestOrderTopLevelDashboardComponents_SortablesSorted(t *testing.T) { assert.Equal(t, "Alpha", out[0].Title, "sortables ordered by title") assert.Equal(t, "Beta", out[1].Title) } + +// TestEventStreamACLNoLeakToUnauthorizedUser (GHSA-228v-wc5r-j8m7) asserts that EventStream +// does not send execution events or output chunks to users who are not allowed to view that action's logs. +func TestEventStreamACLNoLeakToUnauthorizedUser(t *testing.T) { + cfg, lowUser, adminUser := buildViewPermissionTestConfig(t) + ex := executor.DefaultExecutor(cfg) + ex.RebuildActionMap() + api := newServer(ex) + + binding := ex.FindBindingByID("secret_action") + require.NotNil(t, binding, "secret_action binding must exist") + + clientLow, clientAdmin := addEventStreamTestClients(t, api, lowUser, adminUser) + defer removeEventStreamTestClients(api, clientLow, clientAdmin) + + runEventStreamTestExecution(t, ex, cfg, binding, adminUser) + adminEvents := drainEventStreamUntilFinished(clientAdmin.channel, 2*time.Second) + lowEvents := drainEventStreamWithTimeout(clientLow.channel, 50*time.Millisecond) + + assertEventStreamLowUserReceivesNothing(t, lowEvents) + assertEventStreamAdminReceivesSecretActionEvents(t, adminEvents) +} + +func addEventStreamTestClients(t *testing.T, api *oliveTinAPI, lowUser, adminUser *authpublic.AuthenticatedUser) (*streamingClient, *streamingClient) { + t.Helper() + clientLow := &streamingClient{ + channel: make(chan *apiv1.EventStreamResponse, 20), + AuthenticatedUser: lowUser, + } + clientAdmin := &streamingClient{ + channel: make(chan *apiv1.EventStreamResponse, 20), + AuthenticatedUser: adminUser, + } + api.streamingClientsMutex.Lock() + api.streamingClients[clientLow] = struct{}{} + api.streamingClients[clientAdmin] = struct{}{} + api.streamingClientsMutex.Unlock() + return clientLow, clientAdmin +} + +func removeEventStreamTestClients(api *oliveTinAPI, clientLow, clientAdmin *streamingClient) { + api.streamingClientsMutex.Lock() + delete(api.streamingClients, clientLow) + delete(api.streamingClients, clientAdmin) + api.streamingClientsMutex.Unlock() + close(clientLow.channel) + close(clientAdmin.channel) +} + +func runEventStreamTestExecution(t *testing.T, ex *executor.Executor, cfg *config.Config, binding *executor.ActionBinding, adminUser *authpublic.AuthenticatedUser) { + t.Helper() + execReq := &executor.ExecutionRequest{ + Binding: binding, + Arguments: map[string]string{}, + TrackingID: uuid.NewString(), + Cfg: cfg, + AuthenticatedUser: adminUser, + } + wg, _ := ex.ExecRequest(execReq) + wg.Wait() +} + +func drainEventStreamUntilFinished(ch <-chan *apiv1.EventStreamResponse, timeout time.Duration) []*apiv1.EventStreamResponse { + var out []*apiv1.EventStreamResponse + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + ev, finished := recvEventStreamOne(ch, 50*time.Millisecond) + if ev != nil { + out = append(out, ev) + if finished { + return out + } + } + } + return out +} + +func recvEventStreamOne(ch <-chan *apiv1.EventStreamResponse, timeout time.Duration) (*apiv1.EventStreamResponse, bool) { + select { + case ev := <-ch: + return ev, ev.GetExecutionFinished() != nil + case <-time.After(timeout): + return nil, true + } +} + +func drainEventStreamWithTimeout(ch <-chan *apiv1.EventStreamResponse, timeout time.Duration) []*apiv1.EventStreamResponse { + var out []*apiv1.EventStreamResponse + for { + select { + case ev := <-ch: + out = append(out, ev) + case <-time.After(timeout): + return out + } + } +} + +func assertEventStreamLowUserReceivesNothing(t *testing.T, lowEvents []*apiv1.EventStreamResponse) { + t.Helper() + for _, ev := range lowEvents { + assert.Nil(t, ev.GetExecutionStarted(), "low-privilege user must not receive ExecutionStarted") + assert.Nil(t, ev.GetExecutionFinished(), "low-privilege user must not receive ExecutionFinished") + assert.Nil(t, ev.GetOutputChunk(), "low-privilege user must not receive OutputChunk") + } + assert.Empty(t, lowEvents, "low-privilege user with Logs:false must not receive any execution events") +} + +func assertEventStreamAdminReceivesSecretActionEvents(t *testing.T, adminEvents []*apiv1.EventStreamResponse) { + t.Helper() + var gotStarted, gotFinished bool + for _, ev := range adminEvents { + if ev.GetExecutionStarted() != nil { + gotStarted = true + assert.Equal(t, "secret_action", ev.GetExecutionStarted().LogEntry.GetBindingId()) + } + if ev.GetExecutionFinished() != nil { + gotFinished = true + assert.Equal(t, "secret_action", ev.GetExecutionFinished().LogEntry.GetBindingId()) + } + } + assert.True(t, gotStarted, "admin must receive ExecutionStarted for secret_action") + assert.True(t, gotFinished, "admin must receive ExecutionFinished for secret_action") +}