Authorize the SSE stream with short-lived project-scoped tickets

src/main/java/edu/stanford/protege/webprotege/gateway/SecurityConfig.java

@@ -7,6 +7,7 @@
 import org.springframework.security.config.annotation.web.builders.HttpSecurity;
 import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
 import org.springframework.security.config.annotation.web.configuration.WebSecurityCustomizer;
+import org.springframework.http.HttpMethod;
 import org.springframework.security.config.annotation.web.configurers.AbstractHttpConfigurer;
 import org.springframework.security.core.authority.SimpleGrantedAuthority;
 import org.springframework.security.core.authority.mapping.GrantedAuthoritiesMapper;
@@ -63,6 +64,11 @@ public SecurityFilterChain resourceServerFilterChain(HttpSecurity http) throws E
         http.csrf(AbstractHttpConfigurer::disable)
                 .authorizeHttpRequests(auth -> auth
                         .requestMatchers("/wsapps").permitAll()
+                        // The SSE stream cannot be bearer-authenticated (EventSource sends no Authorization
+                        // header); it is guarded instead by the short-lived stream ticket that the controller
+                        // redeems and re-checks for VIEW_PROJECT on every connect. The ticket-issuing endpoint
+                        // (/data/events/ticket) is deliberately left to anyRequest().authenticated() below.
+                        .requestMatchers(HttpMethod.GET, "/data/projects/*/events").permitAll()
                         .anyRequest()
                         .authenticated()
                 );

src/main/java/edu/stanford/protege/webprotege/gateway/sse/HistoryReplaySseCatchUpService.java

@@ -0,0 +1,99 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import edu.stanford.protege.webprotege.common.ProjectId;
+import edu.stanford.protege.webprotege.event.EventTag;
+import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryRequest;
+import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryResponse;
+import edu.stanford.protege.webprotege.ipc.CommandExecutor;
+import edu.stanford.protege.webprotege.ipc.ExecutionContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import javax.annotation.Nullable;
+
+/**
+ * Replays the events a reconnecting client missed by querying the durable event-history service.
+ *
+ * <p>By the time this runs the registry is already buffering live events for the stream (the
+ * controller subscribes first). This service queries history for everything after {@code lastEventId},
+ * hands the whole batch back to the registry as one catch-up frame stamped with the batch's end
+ * sequence, and lets the registry flush the buffered live events — dropping any the replay already
+ * covered — before resuming live delivery. History flattening drops per-event ordinals, so a single
+ * batch replayed with a strictly-greater-than query and monotonic ids is what satisfies "nothing
+ * missed, nothing duplicated".
+ *
+ * <p>Failure modes never kill the stream, so the client's gap detection can recover the hole:
+ * a fresh connection ({@code null} id) has nothing to replay; a non-numeric id is treated as fresh
+ * but still releases the buffer; a failed or empty history query resumes live-only.
+ */
+@Component
+public class HistoryReplaySseCatchUpService implements SseCatchUpService {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(HistoryReplaySseCatchUpService.class);
+
+    /** Dedupe threshold that keeps every buffered event when there is nothing to replay against. */
+    private static final long REPLAY_NOTHING = -1L;
+
+    private final SseStreamRegistry registry;
+
+    private final CommandExecutor<ProjectEventsQueryRequest, ProjectEventsQueryResponse> eventsQueryExecutor;
+
+    public HistoryReplaySseCatchUpService(SseStreamRegistry registry,
+                                          CommandExecutor<ProjectEventsQueryRequest, ProjectEventsQueryResponse> eventsQueryExecutor) {
+        this.registry = registry;
+        this.eventsQueryExecutor = eventsQueryExecutor;
+    }
+
+    @Override
+    public void catchUp(ProjectId projectId,
+                        @Nullable String lastEventId,
+                        SseEmitter emitter,
+                        ExecutionContext executionContext) {
+        if (lastEventId == null) {
+            // Fresh connection: the registry never began buffering, live delivery is already active.
+            return;
+        }
+        Integer since = parseSequence(lastEventId);
+        if (since == null) {
+            // The stream is buffering (lastEventId was present) but the id is unusable: release the
+            // buffer with no replay so the stream resumes live instead of stalling.
+            LOGGER.warn("Ignoring non-numeric Last-Event-ID '{}' for project {}; resuming live-only", lastEventId, projectId.id());
+            registry.completeCatchUp(projectId, emitter, REPLAY_NOTHING, null);
+            return;
+        }
+        try {
+            ProjectEventsQueryResponse response = queryHistory(projectId, since, executionContext);
+            EventTag endTag = (response != null && response.events != null) ? response.events.endTag() : null;
+            if (endTag == null) {
+                LOGGER.warn("Empty history response for project {} since {}; resuming live-only", projectId.id(), since);
+                registry.completeCatchUp(projectId, emitter, since, null);
+                return;
+            }
+            registry.completeCatchUp(projectId, emitter, endTag.getOrdinal(), response);
+        } catch (Exception e) {
+            // Keep the stream alive; the client's gap detection recovers the missed window.
+            LOGGER.error("History replay failed for project {} since {}; resuming live-only", projectId.id(), since, e);
+            registry.completeCatchUp(projectId, emitter, since, null);
+        }
+    }
+
+    private ProjectEventsQueryResponse queryHistory(ProjectId projectId, int since, ExecutionContext executionContext) throws Exception {
+        ProjectEventsQueryRequest request = new ProjectEventsQueryRequest(projectId, EventTag.get(since));
+        return eventsQueryExecutor.execute(request, executionContext).get();
+    }
+
+    @Nullable
+    private static Integer parseSequence(String lastEventId) {
+        String trimmed = lastEventId.trim();
+        if (trimmed.isEmpty()) {
+            return null;
+        }
+        try {
+            return Integer.valueOf(trimmed);
+        } catch (NumberFormatException e) {
+            return null;
+        }
+    }
+}

src/main/java/edu/stanford/protege/webprotege/gateway/sse/InMemoryStreamTicketStore.java

@@ -0,0 +1,74 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import com.google.common.base.Ticker;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import edu.stanford.protege.webprotege.common.ProjectId;
+import edu.stanford.protege.webprotege.common.UserId;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.security.SecureRandom;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Base64;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * In-memory {@link StreamTicketStore} backed by a Guava cache that evicts each ticket a fixed duration
+ * after it is written ({@code expireAfterWrite}). Ticket values are 128 bits of {@link SecureRandom}
+ * rendered URL-safe, so they carry safely in a query string and are infeasible to guess.
+ *
+ * <p>Single-instance only — a ticket lives only in the heap of the gateway that issued it. See
+ * {@link StreamTicketStore} and #307.
+ */
+@Component
+public class InMemoryStreamTicketStore implements StreamTicketStore {
+
+    /** 16 bytes = 128 bits of entropy, per the ticket brief. */
+    private static final int TICKET_BYTES = 16;
+
+    private final SecureRandom secureRandom = new SecureRandom();
+
+    private final Base64.Encoder encoder = Base64.getUrlEncoder().withoutPadding();
+
+    private final Duration ttl;
+
+    private final Cache<String, StreamTicket> tickets;
+
+    @Autowired
+    public InMemoryStreamTicketStore(SseProperties properties) {
+        this(properties.getTicketTtl(), Ticker.systemTicker());
+    }
+
+    /** Package-visible so tests can drive expiry with a fake ticker instead of sleeping out the TTL. */
+    InMemoryStreamTicketStore(Duration ttl, Ticker ticker) {
+        this.ttl = ttl;
+        this.tickets = CacheBuilder.newBuilder()
+                .expireAfterWrite(ttl.toMillis(), TimeUnit.MILLISECONDS)
+                .ticker(ticker)
+                .build();
+    }
+
+    @Override
+    public String issue(UserId userId, ProjectId projectId, String jwt) {
+        String ticket = newTicketValue();
+        tickets.put(ticket, new StreamTicket(userId, projectId, jwt, Instant.now().plus(ttl)));
+        return ticket;
+    }
+
+    @Override
+    public Optional<StreamTicket> redeem(String ticket) {
+        if (ticket == null || ticket.isEmpty()) {
+            return Optional.empty();
+        }
+        return Optional.ofNullable(tickets.getIfPresent(ticket));
+    }
+
+    private String newTicketValue() {
+        byte[] bytes = new byte[TICKET_BYTES];
+        secureRandom.nextBytes(bytes);
+        return encoder.encodeToString(bytes);
+    }
+}

src/main/java/edu/stanford/protege/webprotege/gateway/sse/ProjectEventsSseController.java

@@ -0,0 +1,70 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import edu.stanford.protege.webprotege.common.ProjectId;
+import edu.stanford.protege.webprotege.ipc.ExecutionContext;
+import jakarta.servlet.http.HttpServletResponse;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+/**
+ * Streams project-change events to a single viewer over a long-lived HTTP connection (server-sent events).
+ *
+ * <p>The connection is authorized by a short-lived, project-scoped {@code ticket} query parameter (issued by
+ * {@link StreamTicketController}) rather than a bearer header, because {@code EventSource} cannot set custom
+ * headers. Every connection — including each native auto-reconnect, which re-GETs the same URL — redeems the
+ * ticket afresh via {@link StreamTicketService}, so {@code VIEW_PROJECT} is re-checked on every (re)connect.
+ * Identity comes solely from the server-side ticket record; nothing the request carries is trusted for
+ * authorization. Redemption yields 401 for a missing/unknown/expired/wrong-project ticket and 403 once the
+ * ticket's identity has lost view access.
+ */
+@RestController
+public class ProjectEventsSseController {
+
+    static final String EVENTS_PATH = "/data/projects/{projectId}/events";
+
+    /** nginx-specific header that disables response buffering so events flush immediately. */
+    static final String X_ACCEL_BUFFERING = "X-Accel-Buffering";
+
+    private static final String PROJECT_ID = "projectId";
+
+    private final SseStreamRegistry registry;
+
+    private final SseCatchUpService catchUpService;
+
+    private final StreamTicketService ticketService;
+
+    public ProjectEventsSseController(SseStreamRegistry registry,
+                                      SseCatchUpService catchUpService,
+                                      StreamTicketService ticketService) {
+        this.registry = registry;
+        this.catchUpService = catchUpService;
+        this.ticketService = ticketService;
+    }
+
+    @GetMapping(path = EVENTS_PATH, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public SseEmitter streamEvents(@PathVariable(PROJECT_ID) ProjectId projectId,
+                                   @RequestParam(value = "ticket", required = false) String ticket,
+                                   @RequestParam(value = "lastEventId", required = false) String lastEventIdParam,
+                                   @RequestHeader(value = "Last-Event-ID", required = false) String lastEventIdHeader,
+                                   HttpServletResponse response) {
+        // Redeem before touching the stream: throws 401/403 if the ticket is not a valid, still-authorized
+        // pass for this exact project. The identity/authorization context comes only from the ticket record.
+        ExecutionContext executionContext = ticketService.redeem(ticket, projectId);
+
+        // Browsers send Last-Event-ID only on automatic reconnects; fresh loads use the query param.
+        String lastEventId = (lastEventIdHeader != null) ? lastEventIdHeader : lastEventIdParam;
+
+        response.setHeader(HttpHeaders.CACHE_CONTROL, "no-cache");
+        response.setHeader(X_ACCEL_BUFFERING, "no");
+
+        SseEmitter emitter = registry.subscribe(projectId, lastEventId, executionContext);
+        catchUpService.catchUp(projectId, lastEventId, emitter, executionContext);
+        return emitter;
+    }
+}

src/main/java/edu/stanford/protege/webprotege/gateway/sse/SseCatchUpService.java

@@ -0,0 +1,29 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import edu.stanford.protege.webprotege.common.ProjectId;
+import edu.stanford.protege.webprotege.ipc.ExecutionContext;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+/**
+ * Seam for replaying events a reconnecting client missed while it was disconnected.
+ *
+ * <p>The controller subscribes the emitter to live events first, then calls this service so the
+ * events missed since {@code lastEventId} can be replayed from the durable history ahead of the
+ * live events buffered in the meantime. See {@link HistoryReplaySseCatchUpService}.
+ */
+public interface SseCatchUpService {
+
+    /**
+     * Replay events that occurred after {@code lastEventId} onto {@code emitter}.
+     *
+     * @param projectId        the project whose events are streamed.
+     * @param lastEventId      the last event id the client already received, or {@code null} for a
+     *                         fresh connection with nothing to replay.
+     * @param emitter          the already-subscribed emitter to replay onto.
+     * @param executionContext the identity/authorization context resolved for this connection.
+     */
+    void catchUp(ProjectId projectId,
+                 String lastEventId,
+                 SseEmitter emitter,
+                 ExecutionContext executionContext);
+}

src/main/java/edu/stanford/protege/webprotege/gateway/sse/SseCommandExecutorConfiguration.java

@@ -0,0 +1,23 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryRequest;
+import edu.stanford.protege.webprotege.gateway.websocket.dto.ProjectEventsQueryResponse;
+import edu.stanford.protege.webprotege.ipc.CommandExecutor;
+import edu.stanford.protege.webprotege.ipc.impl.CommandExecutorImpl;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Wires the command executor the SSE catch-up path uses to pull missed events from the durable
+ * event-history service on {@code webprotege.hierarchies.GetProjectEvents}. Declared the same way the
+ * rest of the platform declares {@link CommandExecutor} beans (see the ipc application's
+ * authorization-status executor).
+ */
+@Configuration
+public class SseCommandExecutorConfiguration {
+
+    @Bean
+    CommandExecutor<ProjectEventsQueryRequest, ProjectEventsQueryResponse> projectEventsQueryExecutor() {
+        return new CommandExecutorImpl<>(ProjectEventsQueryResponse.class);
+    }
+}

src/main/java/edu/stanford/protege/webprotege/gateway/sse/SseProperties.java

@@ -0,0 +1,68 @@
+package edu.stanford.protege.webprotege.gateway.sse;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.time.Duration;
+
+/**
+ * Tunables for the server-sent events streaming endpoint.
+ *
+ * <p>The heartbeat interval must stay comfortably below the nginx {@code proxy_read_timeout}
+ * (60s by default) so idle streams are not dropped. The stream timeout closes long-lived
+ * connections so each reconnect becomes a fresh authorization checkpoint.
+ */
+@ConfigurationProperties(prefix = "webprotege.sse")
+public class SseProperties {
+
+    private Duration heartbeatInterval = Duration.ofSeconds(20);
+
+    private Duration streamTimeout = Duration.ofMinutes(30);
+
+    /**
+     * How long a stream ticket stays valid after it is issued. Short by design: a ticket travels in the
+     * stream URL's query string (and so into nginx access logs and browser history), and the window bounds
+     * that exposure. It must be long enough to cover a browser's native reconnect cadence, since the client
+     * reuses the same ticket for the whole TTL. Configurable via {@code webprotege.sse.ticket-ttl} so tests
+     * need not wait out the default.
+     */
+    private Duration ticketTtl = Duration.ofSeconds(120);
+
+    /**
+     * Cap on the live events a reconnecting stream buffers while its history replay is fetched.
+     * Overflow drops the excess and lets the client's gap detection recover; it only guards against
+     * unbounded growth if a history query stalls.
+     */
+    private int catchUpBufferLimit = 1000;
+
+    public Duration getHeartbeatInterval() {
+        return heartbeatInterval;
+    }
+
+    public void setHeartbeatInterval(Duration heartbeatInterval) {
+        this.heartbeatInterval = heartbeatInterval;
+    }
+
+    public Duration getStreamTimeout() {
+        return streamTimeout;
+    }
+
+    public void setStreamTimeout(Duration streamTimeout) {
+        this.streamTimeout = streamTimeout;
+    }
+
+    public int getCatchUpBufferLimit() {
+        return catchUpBufferLimit;
+    }
+
+    public void setCatchUpBufferLimit(int catchUpBufferLimit) {
+        this.catchUpBufferLimit = catchUpBufferLimit;
+    }
+
+    public Duration getTicketTtl() {
+        return ticketTtl;
+    }
+
+    public void setTicketTtl(Duration ticketTtl) {
+        this.ticketTtl = ticketTtl;
+    }
+}