diff --git a/client/internal/approval/broker.go b/client/internal/approval/broker.go index 08eaf72f6..0158283b1 100644 --- a/client/internal/approval/broker.go +++ b/client/internal/approval/broker.go @@ -90,13 +90,15 @@ var ErrDenied = errors.New("approval denied") // EventPublisher is the subset of peer.Status used to emit prompts. type EventPublisher interface { + // PublishEvent reports whether the event reached at least one live + // subscriber. A prompt nobody received is one nobody can answer. PublishEvent( severity proto.SystemEvent_Severity, category proto.SystemEvent_Category, msg string, userMsg string, metadata map[string]string, - ) + ) bool HasEventSubscribers() bool } @@ -171,7 +173,14 @@ func (b *Broker) Request(ctx context.Context, p Prompt) (Decision, error) { if subject == "" { subject = fmt.Sprintf("%s connection requires approval", p.Kind) } - b.pub.PublishEvent(proto.SystemEvent_INFO, proto.SystemEvent_APPROVAL, subject, subject, meta) + // A subscriber whose queue is full drops the event silently. Waiting out the + // timeout for a prompt that never appeared spends the caller's whole + // approval window before denying, and tells the user it timed out rather + // than that nothing ever asked them. + if !b.pub.PublishEvent(proto.SystemEvent_INFO, proto.SystemEvent_APPROVAL, subject, subject, meta) { + log.Warnf("approval request %s (%s) reached no subscriber; denying without waiting", id, p.Kind) + return zero, ErrNoSubscriber + } log.Debugf("approval request %s (%s) emitted: %s", id, p.Kind, subject) timer := time.NewTimer(timeout) diff --git a/client/internal/approval/broker_test.go b/client/internal/approval/broker_test.go index a8e748468..cbbfd6f61 100644 --- a/client/internal/approval/broker_test.go +++ b/client/internal/approval/broker_test.go @@ -20,6 +20,9 @@ type fakePublisher struct { mu sync.Mutex subscribers bool events []*proto.SystemEvent + // dropped makes PublishEvent report that no subscriber received the event, + // standing in for a subscriber whose queue is full. + dropped bool } func (p *fakePublisher) PublishEvent( @@ -28,7 +31,7 @@ func (p *fakePublisher) PublishEvent( msg string, userMsg string, metadata map[string]string, -) { +) bool { p.mu.Lock() p.events = append(p.events, &proto.SystemEvent{ Severity: severity, @@ -37,7 +40,9 @@ func (p *fakePublisher) PublishEvent( UserMessage: userMsg, Metadata: metadata, }) + dropped := p.dropped p.mu.Unlock() + return !dropped } func (p *fakePublisher) HasEventSubscribers() bool { @@ -60,6 +65,25 @@ func (p *fakePublisher) eventCount() int { return len(p.events) } +// A subscriber can exist and still not receive the prompt, because its event +// queue is full and the publisher drops rather than blocks. Waiting out the +// approval window for a dialog that never opened wastes the caller's whole +// timeout and then reports it as one, so the broker refuses straight away. +func TestRequestUndeliveredPromptFailsFast(t *testing.T) { + pub := &fakePublisher{subscribers: true, dropped: true} + b := New(pub) + + start := time.Now() + _, err := b.Request(context.Background(), Prompt{Kind: KindVNC, Subject: "test"}) + assert.ErrorIs(t, err, ErrNoSubscriber) + assert.Less(t, time.Since(start), time.Second, "must not wait out the approval timeout") + + b.mu.Lock() + pending := len(b.pending) + b.mu.Unlock() + assert.Equal(t, 0, pending, "no waiter must be left behind") +} + // TestRequestNoSubscriberFailsClosed is the core fail-closed invariant: // when the UI is not subscribed, the broker must refuse without emitting // an event or arming a waiter. A regression here is a silent bypass. diff --git a/client/internal/auth/sessionwatch/watcher.go b/client/internal/auth/sessionwatch/watcher.go index e685c28d0..3c235b085 100644 --- a/client/internal/auth/sessionwatch/watcher.go +++ b/client/internal/auth/sessionwatch/watcher.go @@ -76,13 +76,15 @@ var ( // owned by sessionwatch, not the caller. type StatusRecorder interface { SetSessionExpiresAt(deadline time.Time) + // PublishEvent reports whether any subscriber received the event; the + // watcher does not act on it. PublishEvent( severity cProto.SystemEvent_Severity, category cProto.SystemEvent_Category, message string, userMessage string, metadata map[string]string, - ) + ) bool } // Watcher observes the latest session deadline and fires two warnings diff --git a/client/internal/auth/sessionwatch/watcher_test.go b/client/internal/auth/sessionwatch/watcher_test.go index 4b49a94b6..0445736c6 100644 --- a/client/internal/auth/sessionwatch/watcher_test.go +++ b/client/internal/auth/sessionwatch/watcher_test.go @@ -64,7 +64,7 @@ func (r *fakeRecorder) PublishEvent( message string, _ string, metadata map[string]string, -) { +) bool { r.mu.Lock() defer r.mu.Unlock() r.events = append(r.events, event{ @@ -74,6 +74,7 @@ func (r *fakeRecorder) PublishEvent( message: message, meta: metadata, }) + return true } func (r *fakeRecorder) snapshot() []event { diff --git a/client/internal/peer/status.go b/client/internal/peer/status.go index f72fc0bc4..6cd8cbc43 100644 --- a/client/internal/peer/status.go +++ b/client/internal/peer/status.go @@ -1306,7 +1306,7 @@ func (d *Status) PublishEvent( msg string, userMsg string, metadata map[string]string, -) { +) bool { event := &proto.SystemEvent{ Id: uuid.New().String(), Severity: severity, @@ -1322,15 +1322,18 @@ func (d *Status) PublishEvent( d.eventQueue.Add(event) + delivered := false for _, stream := range d.eventStreams { select { case stream <- event: + delivered = true default: log.Debugf("event stream buffer full, skipping event: %v", event) } } log.Debugf("event published: %v", event) + return delivered } // SubscribeToEvents returns a new event subscription