security: GHSA-228v-wc5r-j8m7 (HIGH) Unauthorized Action Output Disclosure via EventStream

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 {
+		api.maybeSendExecutionStarted(client, ex, &toRemove)
-		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)
-		}
 	}
-
 	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 {
+		api.maybeSendExecutionFinished(client, ile, &toRemove)
-		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)
-		}
 	}
-
 	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{}
+	entry := api.getValidLogEntryForStreaming(executionTrackingId)
-
+	if entry == nil {
-	for _, client := range api.copyOfStreamingClients() {
+		return
-		select {
+	}
-		case client.channel <- &apiv1.EventStreamResponse{
+	msg := &apiv1.EventStreamResponse{
-			Event: &apiv1.EventStreamResponse_OutputChunk{
+		Event: &apiv1.EventStreamResponse_OutputChunk{
-				OutputChunk: &apiv1.EventOutputChunk{
+			OutputChunk: &apiv1.EventOutputChunk{
-					Output:              string(content),
+				Output:              string(content),
-					ExecutionTrackingId: executionTrackingId,
+				ExecutionTrackingId: executionTrackingId,
-				},
+			},
-			},
+		},
-		}:
+	}
-		default:
+	toRemove := []*streamingClient{}
-			log.Warnf("EventStream: client channel is full, removing client")
+	for _, client := range api.copyOfStreamingClients() {
-			toRemove = append(toRemove, client)
+		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)
 

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")
+}