This is an automated email from the ASF dual-hosted git repository.
sergehuber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new 4d4d73922 Require authentication for GraphQL WebSocket upgrade
(master) (#843)
4d4d73922 is described below
commit 4d4d73922d08de2bf9433f8eea3414a5a414b1cb
Author: Serge Huber <[email protected]>
AuthorDate: Fri Sep 4 11:58:35 2026 +0200
Require authentication for GraphQL WebSocket upgrade (master) (#843)
---
.../fetchers/event/ContextBoundPublisher.java | 133 +++++++++
.../event/EventListenerSubscriptionFetcher.java | 39 ++-
.../unomi/graphql/servlet/GraphQLServlet.java | 147 +++++++++-
.../auth/GraphQLServletSecurityValidator.java | 67 ++++-
.../servlet/websocket/SubscriptionWebSocket.java | 293 +++++++++++++++++--
.../websocket/SubscriptionWebSocketFactory.java | 50 +++-
.../fetchers/event/ContextBoundPublisherTest.java | 151 ++++++++++
.../unomi/graphql/servlet/GraphQLServletTest.java | 264 +++++++++++++++++
.../auth/GraphQLServletSecurityValidatorTest.java | 89 +++++-
.../SubscriptionWebSocketFactoryTest.java | 94 ++++++
.../websocket/SubscriptionWebSocketTest.java | 320 +++++++++++++++++++++
.../src/main/resources/assets/js/index.jsx | 30 +-
.../test/java/org/apache/unomi/itests/AllITs.java | 1 +
.../apache/unomi/itests/CorePersistenceITs.java | 1 +
.../itests/graphql/GraphQLServletSecurityIT.java | 1 +
.../unomi/itests/graphql/GraphQLWebSocketIT.java | 217 +++++++++++++-
.../graphql/socket/out/init-bad-credentials.json | 6 +
.../graphql/socket/out/init-with-credentials.json | 6 +
18 files changed, 1847 insertions(+), 62 deletions(-)
diff --git
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/fetchers/event/ContextBoundPublisher.java
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/fetchers/event/ContextBoundPublisher.java
new file mode 100644
index 000000000..c801e4e55
--- /dev/null
+++
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/fetchers/event/ContextBoundPublisher.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.graphql.fetchers.event;
+
+import org.apache.unomi.api.ExecutionContext;
+import org.apache.unomi.api.security.SecurityService;
+import org.apache.unomi.api.services.ExecutionContextManager;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import javax.security.auth.Subject;
+
+/**
+ * Binds a subscription's own security identity to every delivery made on its
stream.
+ * <p>
+ * A subscription is registered on the caller's thread, where the subject and
execution context are
+ * bound, but its selection set is executed later — once per emitted event, on
whichever thread
+ * produced that event (an event-processing thread shared across tenants).
Without this wrapper the
+ * selection set therefore runs under whatever identity that producer thread
happens to carry, which is
+ * not the subscriber's and may belong to another tenant.
+ * <p>
+ * Wrapping the <em>source</em> publisher is what makes this work: the
framework maps source events to
+ * responses downstream of this point, so binding here covers the
selection-set execution and every
+ * fetcher it reaches. The binding is cleared in a {@code finally} on each
delivery, so no identity is
+ * left behind on a pooled producer thread.
+ */
+public class ContextBoundPublisher<T> implements Publisher<T> {
+
+ private final Publisher<T> delegate;
+ private final Subject subject;
+ private final ExecutionContext executionContext;
+ private final SecurityService securityService;
+ private final ExecutionContextManager executionContextManager;
+
+ public ContextBoundPublisher(final Publisher<T> delegate,
+ final Subject subject,
+ final ExecutionContext executionContext,
+ final SecurityService securityService,
+ final ExecutionContextManager
executionContextManager) {
+ this.delegate = delegate;
+ this.subject = subject;
+ this.executionContext = executionContext;
+ this.securityService = securityService;
+ this.executionContextManager = executionContextManager;
+ }
+
+ @Override
+ public void subscribe(final Subscriber<? super T> subscriber) {
+ delegate.subscribe(new ContextBoundSubscriber(subscriber));
+ }
+
+ private final class ContextBoundSubscriber implements Subscriber<T> {
+
+ private final Subscriber<? super T> delegateSubscriber;
+
+ private ContextBoundSubscriber(final Subscriber<? super T>
delegateSubscriber) {
+ this.delegateSubscriber = delegateSubscriber;
+ }
+
+ @Override
+ public void onSubscribe(final Subscription subscription) {
+ delegateSubscriber.onSubscribe(subscription);
+ }
+
+ @Override
+ public void onNext(final T item) {
+ bind();
+ try {
+ delegateSubscriber.onNext(item);
+ } finally {
+ clear();
+ }
+ }
+
+ @Override
+ public void onError(final Throwable throwable) {
+ bind();
+ try {
+ delegateSubscriber.onError(throwable);
+ } finally {
+ clear();
+ }
+ }
+
+ @Override
+ public void onComplete() {
+ bind();
+ try {
+ delegateSubscriber.onComplete();
+ } finally {
+ clear();
+ }
+ }
+
+ private void bind() {
+ if (securityService != null) {
+ securityService.setCurrentSubject(subject);
+ }
+ if (executionContextManager != null) {
+ executionContextManager.setCurrentContext(executionContext);
+ }
+ }
+
+ private void clear() {
+ // Always unbind: this runs on a shared producer thread, so a
leaked identity would be
+ // inherited by unrelated work scheduled on it afterwards.
+ try {
+ if (securityService != null) {
+ securityService.clearCurrentSubject();
+ }
+ } finally {
+ if (executionContextManager != null) {
+ executionContextManager.setCurrentContext(null);
+ }
+ }
+ }
+ }
+}
diff --git
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/fetchers/event/EventListenerSubscriptionFetcher.java
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/fetchers/event/EventListenerSubscriptionFetcher.java
index 1db0de935..790ce8ff8 100644
---
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/fetchers/event/EventListenerSubscriptionFetcher.java
+++
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/fetchers/event/EventListenerSubscriptionFetcher.java
@@ -17,13 +17,18 @@
package org.apache.unomi.graphql.fetchers.event;
import graphql.schema.DataFetchingEnvironment;
+import org.apache.unomi.api.ExecutionContext;
import org.apache.unomi.api.conditions.Condition;
+import org.apache.unomi.api.security.SecurityService;
+import org.apache.unomi.api.services.ExecutionContextManager;
import org.apache.unomi.graphql.condition.factories.EventConditionFactory;
import org.apache.unomi.graphql.fetchers.BaseDataFetcher;
import org.apache.unomi.graphql.types.input.CDPEventFilterInput;
+import org.apache.unomi.graphql.services.ServiceManager;
import org.apache.unomi.graphql.types.output.CDPEventInterface;
import org.reactivestreams.Publisher;
+import javax.security.auth.Subject;
import java.util.Map;
public class EventListenerSubscriptionFetcher extends
BaseDataFetcher<Publisher<CDPEventInterface>> {
@@ -37,13 +42,43 @@ public class EventListenerSubscriptionFetcher extends
BaseDataFetcher<Publisher<
@Override
public Publisher<CDPEventInterface> get(DataFetchingEnvironment
environment) throws Exception {
Map<String, Object> filterAsMap = environment.getArgument("filter");
+
+ final Publisher<CDPEventInterface> publisher;
if (filterAsMap == null) {
- return eventPublisher.createPublisher();
+ publisher = eventPublisher.createPublisher();
} else {
final CDPEventFilterInput filterInput =
CDPEventFilterInput.fromMap(filterAsMap);
final Condition filterCondition =
EventConditionFactory.get(environment).eventFilterInputCondition(filterInput,
filterAsMap);
- return eventPublisher.createPublisher(filterCondition);
+ publisher = eventPublisher.createPublisher(filterCondition);
+ }
+
+ return bindToSubscriberIdentity(publisher, environment);
+ }
+
+ /**
+ * Captures the identity this subscription is being created under, while
it is still bound to this
+ * thread, and carries it onto every later delivery. Events are emitted
from a producer thread that
+ * carries no identity of this subscriber's, so without this the selection
set would execute under
+ * whatever identity that thread happened to hold.
+ */
+ private Publisher<CDPEventInterface> bindToSubscriberIdentity(
+ final Publisher<CDPEventInterface> publisher, final
DataFetchingEnvironment environment) {
+ final Object context = environment.getContext();
+ if (!(context instanceof ServiceManager)) {
+ return publisher;
+ }
+ final ServiceManager serviceManager = (ServiceManager) context;
+ final SecurityService securityService =
serviceManager.getService(SecurityService.class);
+ final ExecutionContextManager executionContextManager =
serviceManager.getService(ExecutionContextManager.class);
+ if (securityService == null && executionContextManager == null) {
+ return publisher;
}
+
+ final Subject subject = securityService != null ?
securityService.getCurrentSubject() : null;
+ final ExecutionContext executionContext =
+ executionContextManager != null ?
executionContextManager.getCurrentContext() : null;
+
+ return new ContextBoundPublisher<>(publisher, subject,
executionContext, securityService, executionContextManager);
}
}
diff --git
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/GraphQLServlet.java
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/GraphQLServlet.java
index dad1645a0..3443caca8 100644
---
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/GraphQLServlet.java
+++
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/GraphQLServlet.java
@@ -30,7 +30,6 @@ import org.apache.unomi.graphql.services.ServiceManager;
import org.apache.unomi.graphql.servlet.auth.GraphQLServletSecurityValidator;
import org.apache.unomi.graphql.servlet.websocket.SubscriptionWebSocketFactory;
import org.apache.unomi.graphql.utils.GraphQLObjectMapper;
-import org.eclipse.jetty.websocket.servlet.ServletUpgradeRequest;
import org.eclipse.jetty.websocket.servlet.WebSocketServlet;
import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
import org.osgi.service.component.annotations.Component;
@@ -44,7 +43,9 @@ import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
+import java.net.URI;
import java.net.URISyntaxException;
+import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
@@ -98,8 +99,11 @@ public class GraphQLServlet extends WebSocketServlet {
@Override
public void init(ServletConfig config) throws ServletException {
LOGGER.debug("GraphQLServlet initialized");
- super.init(config);
+ // Must precede super.init(): WebSocketServlet.init() calls
configure(), which captures this
+ // validator into the SubscriptionWebSocketFactory. Constructing it
afterwards left the factory -
+ // and therefore every socket - with a null validator, so
connection_init could never authenticate.
this.validator = new GraphQLServletSecurityValidator(tenantService,
securityService, executionContextManager);
+ super.init(config);
}
private WebSocketServletFactory factory;
@@ -108,8 +112,9 @@ public class GraphQLServlet extends WebSocketServlet {
public void configure(WebSocketServletFactory factory) {
LOGGER.debug("GraphQLServlet configured");
this.factory = factory;
- // Wrap the WebSocket creator to handle security context for WebSocket
connections
- SubscriptionWebSocketFactory originalCreator = new
SubscriptionWebSocketFactory(graphQLSchemaUpdater.getGraphQL(), serviceManager);
+ // Wrap the WebSocket creator to bind the authenticated subject
established during upgrade
+ SubscriptionWebSocketFactory originalCreator = new
SubscriptionWebSocketFactory(
+ graphQLSchemaUpdater.getGraphQL(), serviceManager,
securityService, executionContextManager, validator);
factory.setCreator((req, resp) -> {
try {
return originalCreator.createWebSocket(req, resp);
@@ -123,20 +128,132 @@ public class GraphQLServlet extends WebSocketServlet {
@Override
protected void service(HttpServletRequest request, HttpServletResponse
response) throws ServletException, IOException {
LOGGER.debug("GraphQLServlet service called with request: {}",
request.getRequestURI());
- if (factory.isUpgradeRequest(request, response)) {
- try {
- final ServletUpgradeRequest upReq = new
ServletUpgradeRequest(request);
- for (String subProtocol : upReq.getSubProtocols()) {
- if (subProtocol.startsWith("graphql")) {
- response.addHeader("Sec-WebSocket-Protocol",
subProtocol);
- break;
- }
+
+ // HTTP GraphQL (GET/POST/OPTIONS): auth is enforced in doGet/doPost
via validator.validate(...).
+ if (!factory.isUpgradeRequest(request, response)) {
+ serviceNonUpgrade(request, response);
+ return;
+ }
+
+ serviceWebSocketUpgrade(request, response);
+ }
+
+ /**
+ * HTTP GraphQL path. Separated so unit tests can assert upgrade handling
never falls through here.
+ */
+ void serviceNonUpgrade(HttpServletRequest request, HttpServletResponse
response) throws ServletException, IOException {
+ super.service(request, response);
+ }
+
+ /**
+ * WebSocket upgrade path. Order matters for security:
+ * <ol>
+ * <li>Refuse foreign-origin handshakes, then authenticate any
credential the handshake carries
+ * BEFORE {@code acceptWebSocket} (creator reads the thread-local
subject). A handshake that
+ * carries no credential is upgraded unauthenticated - a browser
cannot set headers on it - and
+ * must authenticate through {@code connection_init} before the
socket will do anything.</li>
+ * <li>Call {@code acceptWebSocket} directly — do not call {@code
WebSocketServlet.service()},
+ * which can fall through to {@code doGet}/{@code doPost} when
accept fails and the response
+ * is not committed.</li>
+ * <li>Always clear thread-locals in {@code finally} (covers accept
failures before the creator runs).</li>
+ * </ol>
+ */
+ void serviceWebSocketUpgrade(HttpServletRequest request,
HttpServletResponse response) throws IOException {
+ try {
+ // A WebSocket handshake is not subject to the same-origin policy
and triggers no CORS
+ // preflight, so any page on any origin can open one against this
endpoint. Refuse handshakes
+ // that declare a foreign origin before doing anything else.
+ if (!isOriginAllowed(request)) {
+ LOGGER.warn("Refusing cross-origin WebSocket upgrade from
origin {}", request.getHeader("Origin"));
+ response.sendError(HttpServletResponse.SC_FORBIDDEN,
"Cross-origin WebSocket upgrade refused");
+ return;
+ }
+
+ // Credentials on the upgrade request are the strongest path: the
socket is never created
+ // unauthenticated. They stay mandatory for any client that can
set request headers.
+ //
+ // A browser cannot set headers on a WebSocket handshake, so a
request that carries none is
+ // upgraded in an unauthenticated state instead of being refused.
That socket can do nothing
+ // until it authenticates through connection_init:
SubscriptionWebSocket rejects every other
+ // message until then, and closes the socket if credentials do not
arrive promptly.
+ if (request.getHeader("Authorization") != null &&
!validator.validateWebSocketUpgrade(request, response)) {
+ return;
+ }
+ negotiateGraphqlSubProtocol(request, response);
+ if (!factory.acceptWebSocket(request, response) &&
!response.isCommitted()) {
+ // Upgrade was intended but rejected after auth; never fall
through to HTTP GraphQL.
+ response.sendError(HttpServletResponse.SC_BAD_REQUEST,
"Invalid WebSocket upgrade");
+ }
+ } finally {
+ cleanupSecurityContext();
+ }
+ }
+
+ /**
+ * Accepts a handshake that declares no origin (a non-browser client,
which cannot be driven into
+ * making the request by a hostile page) or one whose origin is this same
host. Anything else is a
+ * page on another origin trying to open a socket here, which is refused.
+ */
+ static boolean isOriginAllowed(HttpServletRequest request) {
+ final String origin = request.getHeader("Origin");
+ if (origin == null || origin.trim().isEmpty()) {
+ return true;
+ }
+ try {
+ final URI originUri = new URI(origin);
+ final String originHost = originUri.getHost();
+ if (originHost == null) {
+ return false;
+ }
+ if (!originHost.equalsIgnoreCase(request.getServerName())) {
+ return false;
+ }
+ int originPort = originUri.getPort();
+ if (originPort == -1) {
+ originPort = "https".equalsIgnoreCase(originUri.getScheme()) ?
443 : 80;
+ }
+ return originPort == request.getServerPort();
+ } catch (URISyntaxException e) {
+ LOGGER.debug("Refusing WebSocket upgrade with unparseable Origin",
e);
+ return false;
+ }
+ }
+
+ /**
+ * Selects the first {@code graphql*} WebSocket subprotocol offered by the
client.
+ * Reads {@code Sec-WebSocket-Protocol} directly so upgrade auth tests do
not need a full Jetty upgrade request.
+ */
+ static void negotiateGraphqlSubProtocol(HttpServletRequest request,
HttpServletResponse response) {
+ Enumeration<String> offered =
request.getHeaders("Sec-WebSocket-Protocol");
+ if (offered == null) {
+ return;
+ }
+ while (offered.hasMoreElements()) {
+ String headerValue = offered.nextElement();
+ if (headerValue == null) {
+ continue;
+ }
+ for (String part : headerValue.split(",")) {
+ String subProtocol = part.trim();
+ if (subProtocol.startsWith("graphql")) {
+ response.addHeader("Sec-WebSocket-Protocol", subProtocol);
+ return;
}
- } catch (URISyntaxException e) {
- throw new RuntimeException(e);
}
}
- super.service(request, response);
+ }
+
+ /**
+ * Package-private wiring for unit tests (avoids full Jetty/OSGi servlet
init).
+ */
+ void bindForTests(WebSocketServletFactory factory,
+ GraphQLServletSecurityValidator validator,
+ SecurityService securityService,
+ ExecutionContextManager executionContextManager) {
+ this.factory = factory;
+ this.validator = validator;
+ this.securityService = securityService;
+ this.executionContextManager = executionContextManager;
}
@Override
diff --git
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/auth/GraphQLServletSecurityValidator.java
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/auth/GraphQLServletSecurityValidator.java
index c391033ce..f9b0ea844 100644
---
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/auth/GraphQLServletSecurityValidator.java
+++
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/auth/GraphQLServletSecurityValidator.java
@@ -38,6 +38,7 @@ import javax.security.auth.login.LoginException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
@@ -65,6 +66,25 @@ public class GraphQLServletSecurityValidator {
this.executionContextManager = executionContextManager;
}
+ /**
+ * Authenticates a WebSocket upgrade. Subscriptions are never public, so
only Basic
+ * (JAAS or tenant private key) is accepted.
+ *
+ * @return true when the caller is authenticated and a security context
was established
+ */
+ public boolean validateWebSocketUpgrade(HttpServletRequest req,
HttpServletResponse res) throws IOException {
+ if (req.getHeader("Authorization") == null) {
+ res.addHeader("WWW-Authenticate", "Basic realm=\"karaf\"");
+ res.sendError(HttpServletResponse.SC_UNAUTHORIZED);
+ return false;
+ }
+ if (isAuthenticatedUser(req)) {
+ return true;
+ }
+ res.sendError(HttpServletResponse.SC_UNAUTHORIZED);
+ return false;
+ }
+
public boolean validate(String query, String operationName,
HttpServletRequest req, HttpServletResponse res) throws IOException {
if (isPublicOperation(query)) {
// For public operations, check API key
@@ -138,15 +158,46 @@ public class GraphQLServletSecurityValidator {
return true;
}
+ /**
+ * Authenticates a Basic credential that did not arrive as a request
header — used by the WebSocket
+ * {@code connection_init} handshake, which is the only way a browser
client can present credentials
+ * (the browser WebSocket API cannot set request headers).
+ * <p>
+ * Deliberately the same credential format and the same verification path
as the header route, so
+ * there is one way to authenticate, not two. No request is involved, so
no tenant header is honoured
+ * here: the caller gets its own tenant's context, never a caller-selected
one.
+ *
+ * @param authorizationValue a {@code Basic <base64>} credential
+ * @return true when the credential authenticated and a security context
was established
+ */
+ public boolean authenticateBasicCredential(String authorizationValue) {
+ return authenticateBasic(authorizationValue, null);
+ }
+
private boolean isAuthenticatedUser(HttpServletRequest req) {
req.setAttribute(AUTHENTICATION_TYPE, HttpServletRequest.BASIC_AUTH);
+ return authenticateBasic(req.getHeader("Authorization"), req);
+ }
- String authHeader = req.getHeader("Authorization");
+ /**
+ * @param req the originating request, or {@code null} when the credential
did not arrive on one
+ * (WebSocket {@code connection_init}); when null, no tenant
header is consulted.
+ */
+ private boolean authenticateBasic(String authHeader, HttpServletRequest
req) {
if (authHeader == null || !authHeader.startsWith("Basic ")) {
return false;
}
- String usernameAndPassword = new
String(Base64.getDecoder().decode(authHeader.substring(6).getBytes()));
+ final String usernameAndPassword;
+ try {
+ usernameAndPassword = new String(
+
Base64.getDecoder().decode(authHeader.substring(6).getBytes(StandardCharsets.UTF_8)),
+ StandardCharsets.UTF_8);
+ } catch (IllegalArgumentException e) {
+ // Malformed Base64 must be treated as an authentication failure
(401), not a 500.
+ LOG.debug("Malformed Basic Authorization header", e);
+ return false;
+ }
int userNameIndex = usernameAndPassword.indexOf(":");
if (userNameIndex == -1) {
return false;
@@ -170,7 +221,9 @@ public class GraphQLServletSecurityValidator {
if (username.length() > 0) {
Tenant tenant = tenantService.getTenantByApiKey(password,
ApiKey.ApiKeyType.PRIVATE);
if (tenant != null && tenant.getItemId().equals(username)) {
- req.setAttribute(REMOTE_USER, username);
+ if (req != null) {
+ req.setAttribute(REMOTE_USER, username);
+ }
// Set the security context for private API key
Subject subject =
securityService.createSubject(tenant.getItemId(), true);
securityService.setCurrentSubject(subject);
@@ -197,12 +250,14 @@ public class GraphQLServletSecurityValidator {
Subject loginSubject = loginContext.getSubject();
boolean success = loginSubject != null;
if (success) {
- req.setAttribute(REMOTE_USER, username);
+ if (req != null) {
+ req.setAttribute(REMOTE_USER, username);
+ }
// Set the security context for JAAS authentication
securityService.setCurrentSubject(loginSubject);
- // Check for tenant ID header
- String tenantId = req.getHeader(UNOMI_TENANT_ID_HEADER);
+ // Check for tenant ID header (only meaningful when the
credential arrived on a request)
+ String tenantId = req != null ?
req.getHeader(UNOMI_TENANT_ID_HEADER) : null;
if (tenantId != null && !tenantId.trim().isEmpty()) {
// Validate tenant exists
Tenant tenant = tenantService.getTenant(tenantId);
diff --git
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocket.java
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocket.java
index 5e20d0305..71299eb19 100644
---
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocket.java
+++
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocket.java
@@ -20,54 +20,156 @@ package org.apache.unomi.graphql.servlet.websocket;
import graphql.ExecutionInput;
import graphql.ExecutionResult;
import graphql.GraphQL;
+import org.apache.unomi.api.ExecutionContext;
+import org.apache.unomi.api.security.SecurityService;
+import org.apache.unomi.api.services.ExecutionContextManager;
import org.apache.unomi.graphql.services.ServiceManager;
+import org.apache.unomi.graphql.servlet.auth.GraphQLServletSecurityValidator;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.WebSocketAdapter;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.security.auth.Subject;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
public class SubscriptionWebSocket extends WebSocketAdapter {
private static final Logger LOGGER =
LoggerFactory.getLogger(SubscriptionWebSocket.class);
- private GraphQL graphQL;
+ private final GraphQL graphQL;
- private ServiceManager serviceManager;
+ private final ServiceManager serviceManager;
+
+ /** Closes a socket that has not authenticated within this window
(milliseconds). */
+ private static final long AUTHENTICATION_DEADLINE_MS = 10_000L;
+
+ private final GraphQLServletSecurityValidator validator;
+
+ /** Set at upgrade for header-authenticated clients, or at connection_init
for browser clients. */
+ private volatile Subject subject;
+
+ private volatile ExecutionContext executionContext;
+
+ /** No operation is executed on this socket until this is true. */
+ private volatile boolean authenticated;
+
+ /** Runs the unauthenticated-socket deadline; owned and stopped by the
factory. */
+ private final ScheduledExecutorService deadlineScheduler;
+
+ /** Makes "authenticate" and "deadline expired" mutually exclusive
outcomes. */
+ private final Object authenticationLock = new Object();
+
+ private volatile ScheduledFuture<?> deadlineTask;
+
+ private boolean deadlineExpired;
+
+ private final SecurityService securityService;
+
+ private final ExecutionContextManager executionContextManager;
private Map<String, ExecutionResultSubscriber> subscriptions = new
HashMap<String, ExecutionResultSubscriber>();
- public SubscriptionWebSocket(GraphQL graphQL, ServiceManager
serviceManager) {
+ public SubscriptionWebSocket(GraphQL graphQL, ServiceManager
serviceManager,
+ Subject subject, ExecutionContext
executionContext,
+ SecurityService securityService,
+ ExecutionContextManager
executionContextManager,
+ GraphQLServletSecurityValidator validator,
+ ScheduledExecutorService deadlineScheduler) {
this.graphQL = graphQL;
+ this.deadlineScheduler = Objects.requireNonNull(deadlineScheduler,
"deadlineScheduler");
this.serviceManager = serviceManager;
+ this.subject = subject;
+ this.executionContext = executionContext;
+ this.securityService = securityService;
+ this.executionContextManager = executionContextManager;
+ this.validator = validator;
+ // A subject supplied here came from an authenticated upgrade;
otherwise the socket starts
+ // unauthenticated and must present credentials through
connection_init.
+ this.authenticated = subject != null;
}
@Override
public void onWebSocketConnect(Session sess) {
LOGGER.info("Opening web socket");
super.onWebSocketConnect(sess);
+ if (!authenticated) {
+ // Bound how long an unauthenticated socket may sit open. The idle
timeout alone is not a
+ // deadline: Jetty resets it on any frame, including ping/pong
control frames that never reach
+ // onWebSocketText, so a client could hold an unauthenticated
socket open just by pinging.
+ // The scheduled task closes the socket at the deadline whatever
the client sends.
+ sess.setIdleTimeout(AUTHENTICATION_DEADLINE_MS);
+ deadlineTask =
deadlineScheduler.schedule(this::expireAuthenticationDeadline,
+ AUTHENTICATION_DEADLINE_MS, TimeUnit.MILLISECONDS);
+ }
}
@Override
public void onWebSocketClose(int statusCode, String reason) {
LOGGER.info("Closing web socket");
+ cancelAuthenticationDeadline();
super.onWebSocketClose(statusCode, reason);
}
+ /**
+ * Runs on the deadline thread when an unauthenticated socket reaches its
deadline. Closes it unless
+ * connection_init already won; the lock makes the two outcomes mutually
exclusive, so a socket is
+ * never authenticated and closed for expiry at the same time.
+ */
+ private void expireAuthenticationDeadline() {
+ synchronized (authenticationLock) {
+ if (authenticated) {
+ return;
+ }
+ deadlineExpired = true;
+ }
+ LOGGER.warn("Closing GraphQL WebSocket that did not authenticate
within {} ms", AUTHENTICATION_DEADLINE_MS);
+ final Session session = getSession();
+ if (session != null && session.isOpen()) {
+ session.close(CLOSE_POLICY_VIOLATION, "Authentication deadline
expired");
+ }
+ }
+
+ private void cancelAuthenticationDeadline() {
+ final ScheduledFuture<?> task = deadlineTask;
+ if (task != null) {
+ task.cancel(false);
+ }
+ }
+
@Override
public void onWebSocketText(String textMessage) {
- LOGGER.info("Got web socket messages {}", textMessage);
+ // Deliberately not logging the message: connection_init carries the
client's credentials.
+ LOGGER.debug("Got web socket message of {} characters", textMessage ==
null ? 0 : textMessage.length());
final GraphQLMessage message = GraphQLMessage.fromJson(textMessage);
if (message == null) {
return;
}
+ // Until the socket has authenticated, connection_init is the only
message that is acted on.
+ // Everything else - including any attempt to start an operation -
closes the socket.
+ if (!authenticated &&
!GraphQLMessage.TYPE_CONNECTION_INIT.equals(message.getType())) {
+ LOGGER.warn("Refusing '{}' on an unauthenticated GraphQL
WebSocket", message.getType());
+ sendMessage(GraphQLMessage.create(message.getId())
+ .type(GraphQLMessage.TYPE_CONNECTION_ERROR)
+ .errors(Collections.singletonList("Not authenticated"))
+ .build());
+ closeConnection(message, CLOSE_POLICY_VIOLATION, "Not
authenticated");
+ return;
+ }
+
switch (message.getType()) {
case GraphQLMessage.TYPE_CONNECTION_INIT:
+ if (!handleConnectionInit(message)) {
+ return;
+ }
sendMessage(GraphQLMessage.connectionAck(message.getId()));
break;
case GraphQLMessage.GQL_START:
@@ -82,9 +184,17 @@ public class SubscriptionWebSocket extends WebSocketAdapter
{
}
}
+ /** WebSocket close codes (RFC 6455). Code 0 is not valid and produces no
client-visible close. */
+ private static final int CLOSE_NORMAL = 1000;
+ private static final int CLOSE_POLICY_VIOLATION = 1008;
+
private void closeConnection(GraphQLMessage message, String reason) {
+ closeConnection(message, CLOSE_NORMAL, reason);
+ }
+
+ private void closeConnection(GraphQLMessage message, int statusCode,
String reason) {
unsubscribe(message);
- getSession().close(0, reason);
+ getSession().close(statusCode, reason);
}
private void sendMessage(GraphQLMessage message) {
@@ -103,36 +213,159 @@ public class SubscriptionWebSocket extends
WebSocketAdapter {
}
}
- private void subscribe(GraphQLMessage message) {
- final Map<String, Object> payload = message.getPayload();
-
- ExecutionInput executionInput = ExecutionInput.newExecutionInput()
- .query((String) payload.get("query"))
- .variables((Map<String, Object>) payload.get("variables"))
- .operationName((String) payload.get("operationName"))
- .context(serviceManager)
- .build();
+ /**
+ * Authenticates the socket from the {@code connection_init} payload,
which is how a browser client
+ * presents credentials (it cannot set request headers on the handshake).
A socket that already
+ * authenticated on the upgrade is left as it is - the payload cannot
replace an established identity.
+ *
+ * @return true when the socket may proceed, false when it has been closed
+ */
+ private boolean handleConnectionInit(GraphQLMessage message) {
+ if (authenticated) {
+ return true;
+ }
+ if (isDeadlineExpired()) {
+ return refuseAfterDeadline(message);
+ }
- ExecutionResult executionResult = this.graphQL.execute(executionInput);
- if (executionResult.getErrors() != null &&
!executionResult.getErrors().isEmpty()) {
- sendMessage(GraphQLMessage.create(message.getId())
- .errors(executionResult.getErrors())
- .build());
- closeConnection(message, "Error executing graphQL query");
- return;
- } else if (!(executionResult.getData() instanceof Publisher)) {
- final String error = "Fetched value should be instance of
Publisher, was: " + executionResult.getClass().getName();
+ final String credential = basicCredentialFrom(message.getPayload());
+ if (credential == null || validator == null ||
!validator.authenticateBasicCredential(credential)) {
+ LOGGER.warn("Refusing GraphQL WebSocket connection_init without a
valid credential");
sendMessage(GraphQLMessage.create(message.getId())
- .errors(Collections.singletonList(error))
+ .type(GraphQLMessage.TYPE_CONNECTION_ERROR)
+ .errors(Collections.singletonList("Not authenticated"))
.build());
- closeConnection(message, error);
- return;
+ closeConnection(message, CLOSE_POLICY_VIOLATION, "Not
authenticated");
+ return false;
+ }
+
+ // The validator establishes the identity on this thread; capture it
onto the socket and unbind,
+ // since this is a shared Jetty IO thread that must not keep carrying
it.
+ try {
+ this.subject = securityService != null ?
securityService.getCurrentSubject() : null;
+ this.executionContext = executionContextManager != null
+ ? executionContextManager.getCurrentContext() : null;
+ } finally {
+ clearThreadSecurityContext();
+ }
+
+ if (this.subject == null) {
+ LOGGER.warn("Refusing GraphQL WebSocket connection_init that
produced no subject");
+ closeConnection(message, CLOSE_POLICY_VIOLATION, "Not
authenticated");
+ return false;
+ }
+
+ synchronized (authenticationLock) {
+ // The deadline may have fired while the credential was being
checked.
+ if (deadlineExpired) {
+ return refuseAfterDeadline(message);
+ }
+ this.authenticated = true;
+ }
+ cancelAuthenticationDeadline();
+ final Session session = getSession();
+ if (session != null) {
+ // Authenticated: drop the short unauthenticated deadline.
+ session.setIdleTimeout(0);
+ }
+ return true;
+ }
+
+ private boolean isDeadlineExpired() {
+ synchronized (authenticationLock) {
+ return deadlineExpired;
+ }
+ }
+
+ /** The deadline task is already closing this socket; a late
connection_init must not resurrect it. */
+ private boolean refuseAfterDeadline(GraphQLMessage message) {
+ LOGGER.warn("Refusing GraphQL WebSocket connection_init that arrived
after the authentication deadline");
+ closeConnection(message, CLOSE_POLICY_VIOLATION, "Not authenticated");
+ return false;
+ }
+
+ /**
+ * Reads a {@code Basic} credential from a connection_init payload. The
same credential format as the
+ * HTTP path, so both routes are verified identically.
+ */
+ private static String basicCredentialFrom(Map<String, Object> payload) {
+ if (payload == null) {
+ return null;
}
+ for (Map.Entry<String, Object> entry : payload.entrySet()) {
+ if (entry.getKey() != null &&
"authorization".equalsIgnoreCase(entry.getKey().trim())
+ && entry.getValue() instanceof String) {
+ return (String) entry.getValue();
+ }
+ }
+ return null;
+ }
- Publisher<ExecutionResult> publisher = executionResult.getData();
- ExecutionResultSubscriber subscriber = new
ExecutionResultSubscriber(message.getId(), getRemote());
- publisher.subscribe(subscriber);
+ private void clearThreadSecurityContext() {
+ try {
+ if (securityService != null) {
+ securityService.clearCurrentSubject();
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error clearing GraphQL WebSocket security context",
e);
+ }
+ try {
+ if (executionContextManager != null) {
+ executionContextManager.setCurrentContext(null);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error clearing GraphQL WebSocket execution context",
e);
+ }
+ }
- subscriptions.put(message.getId(), subscriber);
+ private void subscribe(GraphQLMessage message) {
+ final Map<String, Object> payload = message.getPayload();
+
+ try {
+ securityService.setCurrentSubject(subject);
+ executionContextManager.setCurrentContext(executionContext);
+
+ Map<String, Object> variables = (Map<String, Object>)
payload.get("variables");
+ if (variables == null) {
+ variables = new HashMap<>();
+ }
+
+ ExecutionInput executionInput = ExecutionInput.newExecutionInput()
+ .query((String) payload.get("query"))
+ .variables(variables)
+ .operationName((String) payload.get("operationName"))
+ .context(serviceManager)
+ .build();
+
+ ExecutionResult executionResult =
this.graphQL.execute(executionInput);
+ if (executionResult.getErrors() != null &&
!executionResult.getErrors().isEmpty()) {
+ sendMessage(GraphQLMessage.create(message.getId())
+ .errors(executionResult.getErrors())
+ .build());
+ closeConnection(message, "Error executing graphQL query");
+ return;
+ } else if (!(executionResult.getData() instanceof Publisher)) {
+ Object data = executionResult.getData();
+ final String error = "Fetched value should be instance of
Publisher, was: " + (data == null ? "null" : data.getClass().getName());
+ sendMessage(GraphQLMessage.create(message.getId())
+ .errors(Collections.singletonList(error))
+ .build());
+ closeConnection(message, error);
+ return;
+ }
+
+ Publisher<ExecutionResult> publisher = executionResult.getData();
+ ExecutionResultSubscriber subscriber = new
ExecutionResultSubscriber(message.getId(), getRemote());
+ publisher.subscribe(subscriber);
+
+ subscriptions.put(message.getId(), subscriber);
+ } finally {
+ try {
+ securityService.clearCurrentSubject();
+ executionContextManager.setCurrentContext(null);
+ } catch (Exception e) {
+ LOGGER.error("Error clearing GraphQL WebSocket security
context", e);
+ }
+ }
}
}
diff --git
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocketFactory.java
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocketFactory.java
index c9dde748f..0d0c949da 100644
---
a/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocketFactory.java
+++
b/graphql/cxs-impl/src/main/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocketFactory.java
@@ -18,24 +18,70 @@
package org.apache.unomi.graphql.servlet.websocket;
import graphql.GraphQL;
+import org.apache.unomi.api.ExecutionContext;
+import org.apache.unomi.api.security.SecurityService;
+import org.apache.unomi.api.services.ExecutionContextManager;
import org.apache.unomi.graphql.services.ServiceManager;
+import org.apache.unomi.graphql.servlet.auth.GraphQLServletSecurityValidator;
import org.eclipse.jetty.websocket.server.WebSocketServerFactory;
import org.eclipse.jetty.websocket.servlet.ServletUpgradeRequest;
import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
+import javax.security.auth.Subject;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
public class SubscriptionWebSocketFactory extends WebSocketServerFactory {
private final GraphQL graphQL;
private final ServiceManager serviceManager;
- public SubscriptionWebSocketFactory(GraphQL graphQL, ServiceManager
serviceManager) {
+ private final SecurityService securityService;
+
+ private final ExecutionContextManager executionContextManager;
+
+ private final GraphQLServletSecurityValidator validator;
+
+ /**
+ * Closes sockets that do not authenticate within their deadline. One
daemon thread for all sockets;
+ * stopped with the factory, which {@code WebSocketServlet.destroy()}
stops on undeploy.
+ */
+ private final ScheduledExecutorService authenticationDeadlineScheduler;
+
+ public SubscriptionWebSocketFactory(GraphQL graphQL, ServiceManager
serviceManager,
+ SecurityService securityService,
+ ExecutionContextManager
executionContextManager,
+ GraphQLServletSecurityValidator
validator) {
this.graphQL = graphQL;
this.serviceManager = serviceManager;
+ this.securityService = securityService;
+ this.executionContextManager = executionContextManager;
+ this.validator = validator;
+ this.authenticationDeadlineScheduler =
Executors.newSingleThreadScheduledExecutor(runnable -> {
+ Thread thread = new Thread(runnable,
"graphql-ws-authentication-deadline");
+ thread.setDaemon(true);
+ return thread;
+ });
}
@Override
public Object createWebSocket(ServletUpgradeRequest req,
ServletUpgradeResponse resp) {
- return new SubscriptionWebSocket(graphQL, serviceManager);
+ // A subject here means the upgrade authenticated (header route). Its
absence is not an error:
+ // a browser cannot send credentials on the handshake, so the socket
is created unauthenticated
+ // and must authenticate through connection_init before it can do
anything.
+ Subject subject = securityService.getCurrentSubject();
+ ExecutionContext executionContext = subject != null ?
executionContextManager.getCurrentContext() : null;
+ return new SubscriptionWebSocket(graphQL, serviceManager, subject,
executionContext,
+ securityService, executionContextManager, validator,
authenticationDeadlineScheduler);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ try {
+ super.doStop();
+ } finally {
+ authenticationDeadlineScheduler.shutdownNow();
+ }
}
}
diff --git
a/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/fetchers/event/ContextBoundPublisherTest.java
b/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/fetchers/event/ContextBoundPublisherTest.java
new file mode 100644
index 000000000..089c4e006
--- /dev/null
+++
b/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/fetchers/event/ContextBoundPublisherTest.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.graphql.fetchers.event;
+
+import org.apache.unomi.api.ExecutionContext;
+import org.apache.unomi.api.security.SecurityService;
+import org.apache.unomi.api.services.ExecutionContextManager;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import javax.security.auth.Subject;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.verify;
+
+/**
+ * A subscription's selection set is executed on the thread that produced the
event, not on the thread
+ * that registered the subscription. These tests pin the property that makes
that safe: the subscriber's
+ * own identity is bound for the duration of each delivery, and always unbound
afterwards so it cannot be
+ * inherited by unrelated work on that shared producer thread.
+ */
+@ExtendWith(MockitoExtension.class)
+class ContextBoundPublisherTest {
+
+ @Mock
+ private SecurityService securityService;
+ @Mock
+ private ExecutionContextManager executionContextManager;
+
+ private final Subject subject = new Subject();
+ private final ExecutionContext executionContext = new
ExecutionContext("tenant-a", null, null);
+
+ @Test
+ void delivery_bindsSubscriberIdentityThenUnbinds() {
+ final List<String> observed = new ArrayList<>();
+ final Publisher<String> source = subscriber -> {
+ subscriber.onSubscribe(noopSubscription());
+ subscriber.onNext("event");
+ };
+
+ new ContextBoundPublisher<>(source, subject, executionContext,
securityService, executionContextManager)
+ .subscribe(recordingSubscriber(observed));
+
+ assertEquals(1, observed.size());
+ // Bound before the delivery reached the downstream subscriber...
+ verify(securityService).setCurrentSubject(subject);
+ verify(executionContextManager).setCurrentContext(executionContext);
+ // ...and unbound afterwards.
+ verify(securityService).clearCurrentSubject();
+ verify(executionContextManager).setCurrentContext(null);
+ }
+
+ @Test
+ void delivery_unbindsEvenWhenDownstreamThrows() {
+ final Publisher<String> source = subscriber -> {
+ subscriber.onSubscribe(noopSubscription());
+ try {
+ subscriber.onNext("event");
+ } catch (RuntimeException expected) {
+ // the downstream failure is not what this test asserts on
+ }
+ };
+
+ new ContextBoundPublisher<>(source, subject, executionContext,
securityService, executionContextManager)
+ .subscribe(throwingSubscriber());
+
+ // A leaked identity on a pooled producer thread would be worse than
the failure itself.
+ verify(securityService).clearCurrentSubject();
+ verify(executionContextManager).setCurrentContext(null);
+ }
+
+ @Test
+ void delivery_bindsBeforeDownstreamAndClearsAfter() {
+ final Publisher<String> source = subscriber -> {
+ subscriber.onSubscribe(noopSubscription());
+ subscriber.onNext("event");
+ };
+
+ new ContextBoundPublisher<>(source, subject, executionContext,
securityService, executionContextManager)
+ .subscribe(recordingSubscriber(new ArrayList<>()));
+
+
inOrder(securityService).verify(securityService).setCurrentSubject(subject);
+ inOrder(securityService).verify(securityService).clearCurrentSubject();
+ }
+
+ private Subscription noopSubscription() {
+ return new Subscription() {
+ @Override
+ public void request(long n) {
+ }
+
+ @Override
+ public void cancel() {
+ }
+ };
+ }
+
+ private Subscriber<String> recordingSubscriber(final List<String> sink) {
+ return new BaseSubscriber() {
+ @Override
+ public void onNext(String item) {
+ sink.add(item);
+ }
+ };
+ }
+
+ private Subscriber<String> throwingSubscriber() {
+ return new BaseSubscriber() {
+ @Override
+ public void onNext(String item) {
+ throw new IllegalStateException("downstream failure");
+ }
+ };
+ }
+
+ private abstract static class BaseSubscriber implements Subscriber<String>
{
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ }
+
+ @Override
+ public void onComplete() {
+ }
+ }
+}
diff --git
a/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/GraphQLServletTest.java
b/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/GraphQLServletTest.java
new file mode 100644
index 000000000..ce8f62904
--- /dev/null
+++
b/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/GraphQLServletTest.java
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.unomi.graphql.servlet;
+
+import org.apache.unomi.api.security.SecurityService;
+import org.apache.unomi.api.services.ExecutionContextManager;
+import org.apache.unomi.graphql.servlet.auth.GraphQLServletSecurityValidator;
+import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit coverage for GraphQLServlet WebSocket upgrade ordering: authenticate
first,
+ * acceptWebSocket second, never fall through to HTTP GraphQL, always clear
thread-locals.
+ */
+@ExtendWith(MockitoExtension.class)
+class GraphQLServletTest {
+
+ @Mock
+ private WebSocketServletFactory factory;
+ @Mock
+ private GraphQLServletSecurityValidator validator;
+ @Mock
+ private SecurityService securityService;
+ @Mock
+ private ExecutionContextManager executionContextManager;
+ @Mock
+ private HttpServletRequest request;
+ @Mock
+ private HttpServletResponse response;
+
+ private static final String BASIC_AUTH =
+ "Basic " +
java.util.Base64.getEncoder().encodeToString("user:pass".getBytes());
+
+ private TrackingGraphQLServlet servlet;
+
+ @BeforeEach
+ void setUp() {
+ servlet = new TrackingGraphQLServlet();
+ servlet.bindForTests(factory, validator, securityService,
executionContextManager);
+ }
+
+ @Test
+ void service_nonUpgrade_usesHttpPath_andNeverAcceptsWebSocket() throws
Exception {
+ when(factory.isUpgradeRequest(request, response)).thenReturn(false);
+
+ servlet.service(request, response);
+
+ assertTrue(servlet.nonUpgradeCalled.get());
+ verify(validator, never()).validateWebSocketUpgrade(any(), any());
+ verify(factory, never()).acceptWebSocket(any(), any());
+ verifyNoInteractions(securityService);
+ }
+
+ /** A WebSocket handshake bypasses CORS, so a foreign origin must be
refused outright. */
+ @Test
+ void service_upgrade_crossOrigin_isRefused() throws Exception {
+ when(factory.isUpgradeRequest(request, response)).thenReturn(true);
+
when(request.getHeader("Origin")).thenReturn("https://evil.example.com";);
+ when(request.getServerName()).thenReturn("unomi.example.com");
+
+ servlet.service(request, response);
+
+ verify(response).sendError(eq(HttpServletResponse.SC_FORBIDDEN),
anyString());
+ verify(factory, never()).acceptWebSocket(any(), any());
+ verify(validator, never()).validateWebSocketUpgrade(any(), any());
+ }
+
+ /** Without a credential the upgrade proceeds unauthenticated; the socket
then gates on connection_init. */
+ @Test
+ void service_upgrade_withoutCredential_upgradesUnauthenticated() throws
Exception {
+ when(factory.isUpgradeRequest(request, response)).thenReturn(true);
+ when(request.getHeader("Origin")).thenReturn(null);
+ when(request.getHeader("Authorization")).thenReturn(null);
+
when(request.getHeaders("Sec-WebSocket-Protocol")).thenReturn(Collections.emptyEnumeration());
+ when(factory.acceptWebSocket(request, response)).thenReturn(true);
+
+ servlet.service(request, response);
+
+ verify(validator, never()).validateWebSocketUpgrade(any(), any());
+ verify(factory).acceptWebSocket(request, response);
+ assertFalse(servlet.nonUpgradeCalled.get());
+ }
+
+ @Test
+ void service_upgrade_authRejected_doesNotAccept_clearsContext() throws
Exception {
+ when(factory.isUpgradeRequest(request, response)).thenReturn(true);
+ // No Origin: a non-browser client, which the upgrade accepts (it
cannot be driven by a page).
+ when(request.getHeader("Origin")).thenReturn(null);
+ when(request.getHeader("Authorization")).thenReturn(BASIC_AUTH);
+ when(validator.validateWebSocketUpgrade(request,
response)).thenReturn(false);
+
+ servlet.service(request, response);
+
+ assertFalse(servlet.nonUpgradeCalled.get());
+ verify(factory, never()).acceptWebSocket(any(), any());
+ verify(securityService).clearCurrentSubject();
+ verify(executionContextManager).setCurrentContext(null);
+ verify(response,
never()).sendError(eq(HttpServletResponse.SC_BAD_REQUEST), anyString());
+ }
+
+ @Test
+ void service_upgrade_authAccepted_acceptSucceeds_clearsContext() throws
Exception {
+ when(factory.isUpgradeRequest(request, response)).thenReturn(true);
+ // No Origin: a non-browser client, which the upgrade accepts (it
cannot be driven by a page).
+ when(request.getHeader("Origin")).thenReturn(null);
+ when(request.getHeader("Authorization")).thenReturn(BASIC_AUTH);
+ when(validator.validateWebSocketUpgrade(request,
response)).thenReturn(true);
+
when(request.getHeaders("Sec-WebSocket-Protocol")).thenReturn(Collections.emptyEnumeration());
+ when(factory.acceptWebSocket(request, response)).thenReturn(true);
+
+ servlet.service(request, response);
+
+ assertFalse(servlet.nonUpgradeCalled.get());
+ InOrder order = inOrder(factory, validator, securityService,
executionContextManager);
+ order.verify(factory).isUpgradeRequest(request, response);
+ order.verify(validator).validateWebSocketUpgrade(request, response);
+ order.verify(factory).acceptWebSocket(request, response);
+ order.verify(securityService).clearCurrentSubject();
+ order.verify(executionContextManager).setCurrentContext(null);
+ verify(response,
never()).sendError(eq(HttpServletResponse.SC_BAD_REQUEST), anyString());
+ }
+
+ @Test
+ void
service_upgrade_acceptFailsUncommitted_sends400_neverFallsThroughToHttp()
throws Exception {
+ when(factory.isUpgradeRequest(request, response)).thenReturn(true);
+ // No Origin: a non-browser client, which the upgrade accepts (it
cannot be driven by a page).
+ when(request.getHeader("Origin")).thenReturn(null);
+ when(request.getHeader("Authorization")).thenReturn(BASIC_AUTH);
+ when(validator.validateWebSocketUpgrade(request,
response)).thenReturn(true);
+
when(request.getHeaders("Sec-WebSocket-Protocol")).thenReturn(Collections.emptyEnumeration());
+ when(factory.acceptWebSocket(request, response)).thenReturn(false);
+ when(response.isCommitted()).thenReturn(false);
+
+ servlet.service(request, response);
+
+ assertFalse(servlet.nonUpgradeCalled.get(), "Must not fall through to
HTTP GraphQL after failed accept");
+ verify(response).sendError(HttpServletResponse.SC_BAD_REQUEST,
"Invalid WebSocket upgrade");
+ verify(securityService).clearCurrentSubject();
+ verify(executionContextManager).setCurrentContext(null);
+ }
+
+ @Test
+ void service_upgrade_acceptFailsCommitted_doesNotSendErrorAgain() throws
Exception {
+ when(factory.isUpgradeRequest(request, response)).thenReturn(true);
+ // No Origin: a non-browser client, which the upgrade accepts (it
cannot be driven by a page).
+ when(request.getHeader("Origin")).thenReturn(null);
+ when(request.getHeader("Authorization")).thenReturn(BASIC_AUTH);
+ when(validator.validateWebSocketUpgrade(request,
response)).thenReturn(true);
+
when(request.getHeaders("Sec-WebSocket-Protocol")).thenReturn(Collections.emptyEnumeration());
+ when(factory.acceptWebSocket(request, response)).thenReturn(false);
+ when(response.isCommitted()).thenReturn(true);
+
+ servlet.service(request, response);
+
+ assertFalse(servlet.nonUpgradeCalled.get());
+ verify(response,
never()).sendError(eq(HttpServletResponse.SC_BAD_REQUEST), anyString());
+ verify(securityService).clearCurrentSubject();
+ }
+
+ @Test
+ void service_upgrade_setsGraphqlSubprotocolHeader() throws Exception {
+ when(request.getHeaders("Sec-WebSocket-Protocol"))
+ .thenReturn(enumerationOf("graphql-ws, other"));
+ when(factory.isUpgradeRequest(request, response)).thenReturn(true);
+ // No Origin: a non-browser client, which the upgrade accepts (it
cannot be driven by a page).
+ when(request.getHeader("Origin")).thenReturn(null);
+ when(request.getHeader("Authorization")).thenReturn(BASIC_AUTH);
+ when(validator.validateWebSocketUpgrade(request,
response)).thenReturn(true);
+ when(factory.acceptWebSocket(request, response)).thenReturn(true);
+
+ servlet.service(request, response);
+
+ verify(response).addHeader("Sec-WebSocket-Protocol", "graphql-ws");
+ }
+
+ @Test
+ void negotiateGraphqlSubProtocol_selectsFirstGraphqlToken() {
+ when(request.getHeaders("Sec-WebSocket-Protocol"))
+ .thenReturn(enumerationOf("chat", "graphql-transport-ws,
foo"));
+
+ GraphQLServlet.negotiateGraphqlSubProtocol(request, response);
+
+ verify(response).addHeader("Sec-WebSocket-Protocol",
"graphql-transport-ws");
+ }
+
+ @Test
+ void service_upgrade_clearsContextEvenWhenAcceptThrows() throws Exception {
+ when(factory.isUpgradeRequest(request, response)).thenReturn(true);
+ // No Origin: a non-browser client, which the upgrade accepts (it
cannot be driven by a page).
+ when(request.getHeader("Origin")).thenReturn(null);
+ when(request.getHeader("Authorization")).thenReturn(BASIC_AUTH);
+ when(validator.validateWebSocketUpgrade(request,
response)).thenReturn(true);
+
when(request.getHeaders("Sec-WebSocket-Protocol")).thenReturn(Collections.emptyEnumeration());
+ when(factory.acceptWebSocket(request, response)).thenThrow(new
IOException("boom"));
+
+ try {
+ servlet.service(request, response);
+ fail("Expected IOException");
+ } catch (IOException expected) {
+ // expected
+ }
+
+ assertFalse(servlet.nonUpgradeCalled.get());
+ verify(securityService).clearCurrentSubject();
+ verify(executionContextManager).setCurrentContext(null);
+ }
+
+ private static Enumeration<String> enumerationOf(String... values) {
+ return Collections.enumeration(java.util.Arrays.asList(values));
+ }
+
+ /**
+ * Overrides HTTP fall-through so tests can assert upgrade failures never
reach it.
+ */
+ private static final class TrackingGraphQLServlet extends GraphQLServlet {
+ private final AtomicBoolean nonUpgradeCalled = new
AtomicBoolean(false);
+
+ @Override
+ void serviceNonUpgrade(HttpServletRequest request, HttpServletResponse
response)
+ throws ServletException, IOException {
+ nonUpgradeCalled.set(true);
+ }
+ }
+}
diff --git
a/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/auth/GraphQLServletSecurityValidatorTest.java
b/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/auth/GraphQLServletSecurityValidatorTest.java
index 1023741cd..c8a150e02 100644
---
a/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/auth/GraphQLServletSecurityValidatorTest.java
+++
b/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/auth/GraphQLServletSecurityValidatorTest.java
@@ -52,14 +52,14 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.refEq;
+import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* Covers the JAAS-authenticated tenant resolution branch of {@link
GraphQLServletSecurityValidator},
- * in particular the fallback to {@link ExecutionContext#systemContext()} when
the
- * {@code X-Unomi-Tenant-Id} header does not resolve to a known tenant
(UNOMI-884).
+ * WebSocket upgrade auth (subscriptions are never public), and malformed
Basic handling.
*/
@ExtendWith(MockitoExtension.class)
class GraphQLServletSecurityValidatorTest {
@@ -138,6 +138,91 @@ class GraphQLServletSecurityValidatorTest {
verify(response).sendError(HttpServletResponse.SC_UNAUTHORIZED);
}
+ @Test
+ void validateWebSocketUpgrade_withoutAuthorization_isRejected() throws
IOException {
+ when(request.getHeader("Authorization")).thenReturn(null);
+
+ boolean authenticated = validator.validateWebSocketUpgrade(request,
response);
+
+ assertFalse(authenticated);
+ verify(response).sendError(HttpServletResponse.SC_UNAUTHORIZED);
+ verify(securityService, never()).setCurrentSubject(any());
+ }
+
+ @Test
+ void validateWebSocketUpgrade_withBasicAuth_isAccepted() throws
IOException {
+ when(request.getHeader("Authorization")).thenReturn(BASIC_AUTH);
+ when(tenantService.getTenantByApiKey(any(),
eq(ApiKey.ApiKeyType.PRIVATE))).thenReturn(null);
+
+ boolean authenticated = validator.validateWebSocketUpgrade(request,
response);
+
+ assertTrue(authenticated);
+ verify(securityService).setCurrentSubject(any(Subject.class));
+ verify(response, never()).sendError(any(Integer.class));
+ }
+
+ @Test
+ void validateWebSocketUpgrade_rejectsPublicApiKeyOnly() throws IOException
{
+ // No Authorization header — public API key alone must not open
subscriptions.
+ // validateWebSocketUpgrade never reads X-Unomi-Api-Key; stub
documents the scenario.
+ when(request.getHeader("Authorization")).thenReturn(null);
+
lenient().when(request.getHeader("X-Unomi-Api-Key")).thenReturn("public-api-key");
+
+ boolean authenticated = validator.validateWebSocketUpgrade(request,
response);
+
+ assertFalse(authenticated);
+ verify(response).sendError(HttpServletResponse.SC_UNAUTHORIZED);
+ }
+
+ @Test
+ void validateWebSocketUpgrade_withMalformedBasic_isRejected() throws
IOException {
+ when(request.getHeader("Authorization")).thenReturn("Basic !!!");
+
+ boolean authenticated = validator.validateWebSocketUpgrade(request,
response);
+
+ assertFalse(authenticated);
+ verify(response).sendError(HttpServletResponse.SC_UNAUTHORIZED);
+ verify(securityService, never()).setCurrentSubject(any());
+ }
+
+ @Test
+ void validate_subscriptionQuery_withoutAuthorization_isRejected() throws
IOException {
+ // HTTP contrast: subscriptions are never public — same invariant the
WS upgrade must enforce.
+ when(request.getHeader("Authorization")).thenReturn(null);
+
+ boolean authenticated = validator.validate(
+ "subscription { eventListener { id } }", null, request,
response);
+
+ assertFalse(authenticated);
+ verify(response).sendError(HttpServletResponse.SC_UNAUTHORIZED);
+ verify(securityService, never()).setCurrentSubject(any());
+ }
+
+ @Test
+ void validate_subscriptionQuery_rejectsPublicApiKeyOnly() throws
IOException {
+ // Subscriptions are never public: public API key must not
authenticate a subscription query.
+ when(request.getHeader("Authorization")).thenReturn(null);
+
lenient().when(request.getHeader("X-Unomi-Api-Key")).thenReturn("public-api-key");
+
+ boolean authenticated = validator.validate(
+ "subscription { eventListener { id } }", null, request,
response);
+
+ assertFalse(authenticated);
+ verify(response).sendError(HttpServletResponse.SC_UNAUTHORIZED);
+ verify(securityService, never()).setCurrentSubject(any());
+ }
+
+ @Test
+ void validate_withMalformedBasic_isRejected() throws IOException {
+ when(request.getHeader("Authorization")).thenReturn("Basic
not-valid-base64");
+
+ boolean authenticated = validator.validate(null, null, request,
response);
+
+ assertFalse(authenticated);
+ verify(response).sendError(HttpServletResponse.SC_UNAUTHORIZED);
+ verify(securityService, never()).setCurrentSubject(any());
+ }
+
/**
* An unset {@code org.apache.unomi.security.root.password} resolves to
the empty string, which
* {@code PropertiesLoginModule} accepts as the shipped administrator's
password (UNOMI-974).
diff --git
a/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocketFactoryTest.java
b/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocketFactoryTest.java
new file mode 100644
index 000000000..84e58b1ca
--- /dev/null
+++
b/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocketFactoryTest.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.unomi.graphql.servlet.websocket;
+
+import graphql.GraphQL;
+import org.apache.unomi.api.ExecutionContext;
+import org.apache.unomi.api.security.SecurityService;
+import org.apache.unomi.api.services.ExecutionContextManager;
+import org.apache.unomi.graphql.services.ServiceManager;
+import org.apache.unomi.graphql.servlet.auth.GraphQLServletSecurityValidator;
+import org.eclipse.jetty.websocket.servlet.ServletUpgradeRequest;
+import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import javax.security.auth.Subject;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class SubscriptionWebSocketFactoryTest {
+
+ @Mock
+ private GraphQL graphQL;
+ @Mock
+ private ServiceManager serviceManager;
+ @Mock
+ private SecurityService securityService;
+ @Mock
+ private ExecutionContextManager executionContextManager;
+ @Mock
+ private ServletUpgradeRequest upgradeRequest;
+ @Mock
+ private ServletUpgradeResponse upgradeResponse;
+
+ @Mock
+ private GraphQLServletSecurityValidator validator;
+
+ private SubscriptionWebSocketFactory factory;
+
+ @BeforeEach
+ void setUp() {
+ factory = new SubscriptionWebSocketFactory(graphQL, serviceManager,
securityService, executionContextManager, validator);
+ }
+
+ @Test
+ void createWebSocket_withoutSubject_returnsUnauthenticatedSocket() {
+ when(securityService.getCurrentSubject()).thenReturn(null);
+
+ Object socket = factory.createWebSocket(upgradeRequest,
upgradeResponse);
+
+ // A browser cannot authenticate on the handshake, so the socket is
created unauthenticated
+ // instead of refused; it can do nothing until it authenticates
through connection_init.
+ assertNotNull(socket);
+ verify(upgradeResponse, never()).setStatusCode(401);
+ }
+
+ @Test
+ void createWebSocket_withSubject_returnsSubscriptionWebSocket() {
+ Subject subject = new Subject();
+ ExecutionContext context = new ExecutionContext("tenant-a", null,
null);
+ when(securityService.getCurrentSubject()).thenReturn(subject);
+ when(executionContextManager.getCurrentContext()).thenReturn(context);
+
+ Object socket = factory.createWebSocket(upgradeRequest,
upgradeResponse);
+
+ assertNotNull(socket);
+ assertTrue(socket instanceof SubscriptionWebSocket);
+ verify(upgradeResponse, never()).setStatusCode(401);
+ }
+}
diff --git
a/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocketTest.java
b/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocketTest.java
new file mode 100644
index 000000000..c874f2a7a
--- /dev/null
+++
b/graphql/cxs-impl/src/test/java/org/apache/unomi/graphql/servlet/websocket/SubscriptionWebSocketTest.java
@@ -0,0 +1,320 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.unomi.graphql.servlet.websocket;
+
+import graphql.ExecutionInput;
+import graphql.ExecutionResult;
+import graphql.GraphQL;
+import graphql.GraphQLError;
+import io.reactivex.Flowable;
+import org.apache.unomi.api.ExecutionContext;
+import org.apache.unomi.api.security.SecurityService;
+import org.apache.unomi.api.services.ExecutionContextManager;
+import org.apache.unomi.graphql.services.ServiceManager;
+import org.apache.unomi.graphql.servlet.auth.GraphQLServletSecurityValidator;
+import org.eclipse.jetty.websocket.api.RemoteEndpoint;
+import org.eclipse.jetty.websocket.api.Session;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.reactivestreams.Publisher;
+
+import javax.security.auth.Subject;
+import java.util.Collections;
+import java.util.Map;
+import static org.mockito.Mockito.lenient;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Regression coverage for authenticated subscription execution:
+ * subject/context must be bound for {@code graphQL.execute}, null/omitted
+ * {@code variables} must not NPE inside GraphQL, and thread-locals must be
cleared afterward.
+ */
+@ExtendWith(MockitoExtension.class)
+class SubscriptionWebSocketTest {
+
+ @Mock
+ private GraphQL graphQL;
+ @Mock
+ private ServiceManager serviceManager;
+ @Mock
+ private SecurityService securityService;
+ @Mock
+ private ExecutionContextManager executionContextManager;
+ @Mock
+ private Session session;
+ @Mock
+ private RemoteEndpoint remote;
+ @Mock
+ private ExecutionResult executionResult;
+ @Mock
+ private GraphQLError graphQLError;
+
+ private final Subject subject = new Subject();
+ private final ExecutionContext executionContext = new
ExecutionContext("test-tenant", null, null);
+
+ @Mock
+ private GraphQLServletSecurityValidator validator;
+ @Mock
+ private ScheduledExecutorService deadlineScheduler;
+
+ private SubscriptionWebSocket socket;
+
+ @BeforeEach
+ void setUp() {
+ socket = new SubscriptionWebSocket(graphQL, serviceManager, subject,
executionContext,
+ securityService, executionContextManager, validator,
deadlineScheduler);
+ // Not every test sends a frame; keep the shared stub from tripping
strict-stubs checks.
+ lenient().when(session.getRemote()).thenReturn(remote);
+ socket.onWebSocketConnect(session);
+ }
+
+ /** An unauthenticated socket (the browser path) must not execute anything
before it authenticates. */
+ @Test
+ void unauthenticated_startIsRefusedAndSocketClosed() {
+ SubscriptionWebSocket unauth = new SubscriptionWebSocket(graphQL,
serviceManager, null, null,
+ securityService, executionContextManager, validator,
deadlineScheduler);
+ unauth.onWebSocketConnect(session);
+
+ unauth.onWebSocketText(startMessage("\"variables\":null"));
+
+ verifyNoInteractions(graphQL);
+ // A valid WebSocket close code, not 0: code 0 produces no
client-visible close frame.
+ verify(session).close(eq(1008), anyString());
+ }
+
+ @Test
+ void unauthenticated_connectionInitWithoutCredential_isRefused() {
+ SubscriptionWebSocket unauth = new SubscriptionWebSocket(graphQL,
serviceManager, null, null,
+ securityService, executionContextManager, validator,
deadlineScheduler);
+ unauth.onWebSocketConnect(session);
+
+ unauth.onWebSocketText("{\"type\":\"connection_init\",\"id\":\"1\"}");
+
+ verify(validator, never()).authenticateBasicCredential(anyString());
+ verify(session).close(eq(1008), anyString());
+ }
+
+ @Test
+ void unauthenticated_connectionInitWithBadCredential_isRefused() {
+
when(validator.authenticateBasicCredential(anyString())).thenReturn(false);
+ SubscriptionWebSocket unauth = new SubscriptionWebSocket(graphQL,
serviceManager, null, null,
+ securityService, executionContextManager, validator,
deadlineScheduler);
+ unauth.onWebSocketConnect(session);
+
+ unauth.onWebSocketText("{\"type\":\"connection_init\",\"id\":\"1\","
+ + "\"payload\":{\"Authorization\":\"Basic
Ym9ndXM6Ym9ndXM=\"}}");
+
+ verify(session).close(eq(1008), anyString());
+ }
+
+ /** A valid connection_init credential authenticates the socket and lifts
the unauthenticated deadline. */
+ @Test
+ void
unauthenticated_connectionInitWithValidCredential_authenticatesSocket() {
+
when(validator.authenticateBasicCredential(anyString())).thenReturn(true);
+ when(securityService.getCurrentSubject()).thenReturn(subject);
+
when(executionContextManager.getCurrentContext()).thenReturn(executionContext);
+ SubscriptionWebSocket unauth = new SubscriptionWebSocket(graphQL,
serviceManager, null, null,
+ securityService, executionContextManager, validator,
deadlineScheduler);
+ unauth.onWebSocketConnect(session);
+
+ unauth.onWebSocketText("{\"type\":\"connection_init\",\"id\":\"1\","
+ + "\"payload\":{\"Authorization\":\"Basic dXNlcjpwYXNz\"}}");
+
+ // Identity captured onto the socket, and not left bound to this
shared IO thread.
+ verify(securityService).clearCurrentSubject();
+ verify(session, never()).close(anyInt(), anyString());
+ verify(session).setIdleTimeout(0);
+ }
+
+ @Test
+ void subscribe_bindsSubjectThenClearsThreadLocals() {
+ stubSuccessfulPublisherExecute();
+
+ socket.onWebSocketText(startMessage("\"variables\":null"));
+
+ InOrder order = inOrder(securityService, executionContextManager);
+ order.verify(securityService).setCurrentSubject(subject);
+
order.verify(executionContextManager).setCurrentContext(executionContext);
+ order.verify(securityService).clearCurrentSubject();
+ order.verify(executionContextManager).setCurrentContext(null);
+ }
+
+ @Test
+ void subscribe_nullVariables_passesEmptyMapToExecute() {
+ // Reporter used omitted/null variables to prove unauthenticated
traffic reached
+ // graphQL.execute (NPE: "variables map can't be null"). Authenticated
path must
+ // normalize null → empty map before execute.
+ stubSuccessfulPublisherExecute();
+
+ socket.onWebSocketText(startMessage("\"variables\":null"));
+
+ ArgumentCaptor<ExecutionInput> input =
ArgumentCaptor.forClass(ExecutionInput.class);
+ verify(graphQL).execute(input.capture());
+ Map<String, Object> variables = input.getValue().getVariables();
+ assertNotNull(variables);
+ assertTrue(variables.isEmpty());
+ }
+
+ @Test
+ void subscribe_omittedVariables_passesEmptyMapToExecute() {
+ stubSuccessfulPublisherExecute();
+
+ socket.onWebSocketText("{"
+ + "\"id\":\"1\","
+ + "\"type\":\"start\","
+ + "\"payload\":{"
+ + "\"query\":\"subscription { eventListener { id } }\""
+ + "}}");
+
+ ArgumentCaptor<ExecutionInput> input =
ArgumentCaptor.forClass(ExecutionInput.class);
+ verify(graphQL).execute(input.capture());
+ Map<String, Object> variables = input.getValue().getVariables();
+ assertNotNull(variables);
+ assertTrue(variables.isEmpty());
+ }
+
+ @Test
+ void subscribe_clearsThreadLocalsEvenWhenExecutionFails() {
+
when(executionResult.getErrors()).thenReturn(Collections.singletonList(graphQLError));
+
when(graphQL.execute(any(ExecutionInput.class))).thenReturn(executionResult);
+
+ socket.onWebSocketText(startMessage(null));
+
+ verify(securityService).setCurrentSubject(subject);
+ verify(executionContextManager).setCurrentContext(executionContext);
+ verify(securityService).clearCurrentSubject();
+ verify(executionContextManager).setCurrentContext(null);
+ verify(session).close(anyInt(), anyString());
+ }
+
+ private void stubSuccessfulPublisherExecute() {
+ Publisher<ExecutionResult> publisher = Flowable.never();
+ when(executionResult.getErrors()).thenReturn(Collections.emptyList());
+ when(executionResult.getData()).thenReturn(publisher);
+
when(graphQL.execute(any(ExecutionInput.class))).thenReturn(executionResult);
+ }
+
+ private static String startMessage(String variablesField) {
+ StringBuilder payload = new StringBuilder();
+ payload.append("{\"id\":\"1\",\"type\":\"start\",\"payload\":{");
+ payload.append("\"query\":\"subscription { eventListener { id } }\"");
+ if (variablesField != null) {
+ payload.append(',').append(variablesField);
+ payload.append(",\"operationName\":null");
+ }
+ payload.append("}}");
+ return payload.toString();
+ }
+
+ /** The deadline is a scheduled close, not just an idle timeout: pinging
cannot keep an unauthenticated socket open. */
+ @Test
+ void unauthenticated_socketSchedulesHardAuthenticationDeadline() {
+ SubscriptionWebSocket unauth = new SubscriptionWebSocket(graphQL,
serviceManager, null, null,
+ securityService, executionContextManager, validator,
deadlineScheduler);
+
+ unauth.onWebSocketConnect(session);
+
+ verify(session).setIdleTimeout(10_000L);
+ verify(deadlineScheduler).schedule(any(Runnable.class), eq(10_000L),
eq(TimeUnit.MILLISECONDS));
+ }
+
+ @Test
+ void authenticatedAtUpgrade_schedulesNoDeadline() {
+ // setUp's socket authenticated on the upgrade.
+ verify(deadlineScheduler, never()).schedule(any(Runnable.class),
anyLong(), any(TimeUnit.class));
+ verify(session, never()).setIdleTimeout(anyLong());
+ }
+
+ @Test
+ void deadlineFiring_closesUnauthenticatedSocketWithPolicyViolation() {
+ when(session.isOpen()).thenReturn(true);
+ SubscriptionWebSocket unauth = new SubscriptionWebSocket(graphQL,
serviceManager, null, null,
+ securityService, executionContextManager, validator,
deadlineScheduler);
+ unauth.onWebSocketConnect(session);
+ ArgumentCaptor<Runnable> deadline =
ArgumentCaptor.forClass(Runnable.class);
+ verify(deadlineScheduler).schedule(deadline.capture(), eq(10_000L),
eq(TimeUnit.MILLISECONDS));
+
+ deadline.getValue().run();
+
+ verify(session).close(eq(1008), anyString());
+ }
+
+ /** Authenticating cancels the deadline, and a deadline that fires late
must not close an authenticated socket. */
+ @Test
+ void deadlineFiring_afterAuthentication_doesNotCloseSocket() {
+
when(validator.authenticateBasicCredential(anyString())).thenReturn(true);
+ when(securityService.getCurrentSubject()).thenReturn(subject);
+
when(executionContextManager.getCurrentContext()).thenReturn(executionContext);
+ ScheduledFuture<?> task = mock(ScheduledFuture.class);
+ doReturn(task).when(deadlineScheduler).schedule(any(Runnable.class),
eq(10_000L), eq(TimeUnit.MILLISECONDS));
+ SubscriptionWebSocket unauth = new SubscriptionWebSocket(graphQL,
serviceManager, null, null,
+ securityService, executionContextManager, validator,
deadlineScheduler);
+ unauth.onWebSocketConnect(session);
+ ArgumentCaptor<Runnable> deadline =
ArgumentCaptor.forClass(Runnable.class);
+ verify(deadlineScheduler).schedule(deadline.capture(), eq(10_000L),
eq(TimeUnit.MILLISECONDS));
+
+ unauth.onWebSocketText("{\"type\":\"connection_init\",\"id\":\"1\","
+ + "\"payload\":{\"Authorization\":\"Basic dXNlcjpwYXNz\"}}");
+ deadline.getValue().run();
+
+ verify(task).cancel(false);
+ verify(session, never()).close(anyInt(), anyString());
+ }
+
+ /** Once the deadline has fired, a late connection_init cannot resurrect
the socket. */
+ @Test
+ void connectionInit_afterDeadlineExpired_isRefused() {
+ when(session.isOpen()).thenReturn(true);
+ SubscriptionWebSocket unauth = new SubscriptionWebSocket(graphQL,
serviceManager, null, null,
+ securityService, executionContextManager, validator,
deadlineScheduler);
+ unauth.onWebSocketConnect(session);
+ ArgumentCaptor<Runnable> deadline =
ArgumentCaptor.forClass(Runnable.class);
+ verify(deadlineScheduler).schedule(deadline.capture(), eq(10_000L),
eq(TimeUnit.MILLISECONDS));
+ deadline.getValue().run();
+
+ unauth.onWebSocketText("{\"type\":\"connection_init\",\"id\":\"1\","
+ + "\"payload\":{\"Authorization\":\"Basic dXNlcjpwYXNz\"}}");
+
+ verify(validator, never()).authenticateBasicCredential(anyString());
+ verify(session, atLeastOnce()).close(eq(1008), anyString());
+ verifyNoInteractions(graphQL);
+ }
+}
diff --git a/graphql/graphql-ui/src/main/resources/assets/js/index.jsx
b/graphql/graphql-ui/src/main/resources/assets/js/index.jsx
index a0869796f..a1de4467e 100644
--- a/graphql/graphql-ui/src/main/resources/assets/js/index.jsx
+++ b/graphql/graphql-ui/src/main/resources/assets/js/index.jsx
@@ -33,11 +33,37 @@ function graphqlWsUrl() {
return protocol + '//' + window.location.host + '/graphql';
}
+// The browser WebSocket API cannot set request headers on the handshake, so
the server authenticates a
+// subscription from the connection_init payload instead. GraphiQL hands the
live "Headers" tab content
+// to the fetcher on every request, so capture it here and reuse its
Authorization as the WebSocket
+// connection parameters: HTTP and WebSocket then use the same credential, and
nothing is persisted.
+let latestHeaders = null;
+
+function authorizationHeader() {
+ if (!latestHeaders) {
+ return null;
+ }
+ const key = Object.keys(latestHeaders).find((name) => name.toLowerCase()
=== 'authorization');
+ return key && latestHeaders[key] ? latestHeaders[key] : null;
+}
+
function createFetcher() {
- return createGraphiQLFetcher({
+ const fetcher = createGraphiQLFetcher({
url: graphqlHttpUrl(),
- wsClient: createClient({ url: graphqlWsUrl() }),
+ wsClient: createClient({
+ url: graphqlWsUrl(),
+ // Evaluated on each (re)connect, and sent as the connection_init
payload.
+ connectionParams: () => {
+ const authorization = authorizationHeader();
+ return authorization ? { Authorization: authorization } : {};
+ },
+ }),
});
+
+ return (graphQLParams, opts) => {
+ latestHeaders = (opts && opts.headers) || null;
+ return fetcher(graphQLParams, opts);
+ };
}
function QueryPlayground() {
diff --git a/itests/src/test/java/org/apache/unomi/itests/AllITs.java
b/itests/src/test/java/org/apache/unomi/itests/AllITs.java
index 446c4b559..89609d825 100644
--- a/itests/src/test/java/org/apache/unomi/itests/AllITs.java
+++ b/itests/src/test/java/org/apache/unomi/itests/AllITs.java
@@ -66,6 +66,7 @@ import org.junit.runners.Suite.SuiteClasses;
GraphQLProfileIT.class,
GraphQLProfilePropertiesIT.class,
GraphQLSegmentIT.class,
+ GraphQLServletSecurityIT.class,
GraphQLWebSocketIT.class,
JSONSchemaIT.class,
GraphQLProfileAliasesIT.class,
diff --git
a/itests/src/test/java/org/apache/unomi/itests/CorePersistenceITs.java
b/itests/src/test/java/org/apache/unomi/itests/CorePersistenceITs.java
index f4634d1f4..c279eeef7 100644
--- a/itests/src/test/java/org/apache/unomi/itests/CorePersistenceITs.java
+++ b/itests/src/test/java/org/apache/unomi/itests/CorePersistenceITs.java
@@ -67,6 +67,7 @@ import org.junit.runners.Suite.SuiteClasses;
GraphQLProfileIT.class,
GraphQLProfilePropertiesIT.class,
GraphQLSegmentIT.class,
+ GraphQLServletSecurityIT.class,
GraphQLWebSocketIT.class,
JSONSchemaIT.class,
GraphQLProfileAliasesIT.class,
diff --git
a/itests/src/test/java/org/apache/unomi/itests/graphql/GraphQLServletSecurityIT.java
b/itests/src/test/java/org/apache/unomi/itests/graphql/GraphQLServletSecurityIT.java
index 5fbb58dae..4288ceaaf 100644
---
a/itests/src/test/java/org/apache/unomi/itests/graphql/GraphQLServletSecurityIT.java
+++
b/itests/src/test/java/org/apache/unomi/itests/graphql/GraphQLServletSecurityIT.java
@@ -78,6 +78,7 @@ public class GraphQLServletSecurityIT extends BaseGraphQLIT {
@Test
public void testAnonymousSubscriptionRequest() throws Exception {
+ // HTTP contrast to the WebSocket finding: anonymous subscription POST
must stay 401.
try (CloseableHttpResponse response =
postAnonymous("graphql/security/subscribe.json")) {
Assert.assertEquals(401, response.getStatusLine().getStatusCode());
diff --git
a/itests/src/test/java/org/apache/unomi/itests/graphql/GraphQLWebSocketIT.java
b/itests/src/test/java/org/apache/unomi/itests/graphql/GraphQLWebSocketIT.java
index bcef7ebc0..b0160cb78 100644
---
a/itests/src/test/java/org/apache/unomi/itests/graphql/GraphQLWebSocketIT.java
+++
b/itests/src/test/java/org/apache/unomi/itests/graphql/GraphQLWebSocketIT.java
@@ -23,6 +23,7 @@ import io.reactivex.ObservableEmitter;
import io.reactivex.subscribers.DefaultSubscriber;
import org.eclipse.jetty.websocket.api.RemoteEndpoint;
import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.UpgradeException;
import org.eclipse.jetty.websocket.api.WebSocketAdapter;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;
@@ -32,12 +33,28 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.URI;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Base64;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+/**
+ * End-to-end GraphQL WebSocket auth regression.
+ * <p>
+ * Maps to the reported unauthenticated-subscription scenarios:
+ * <ul>
+ * <li>Upgrade with no credentials → HTTP 401 (not 101)</li>
+ * <li>Upgrade with public API key only → HTTP 401 (subscriptions are never
public)</li>
+ * <li>Upgrade with wrong Basic password → HTTP 401</li>
+ * <li>Upgrade with malformed Basic → HTTP 401</li>
+ * <li>Upgrade with valid JAAS / private key → 101, then connection_init /
start work</li>
+ * </ul>
+ * Unauthenticated clients must never reach {@code connection_ack} or {@code
GQL_START}.
+ */
public class GraphQLWebSocketIT extends BaseGraphQLIT {
private final static Logger LOGGER =
LoggerFactory.getLogger(GraphQLWebSocketIT.class);
@@ -52,9 +69,10 @@ public class GraphQLWebSocketIT extends BaseGraphQLIT {
URI echoUri = new URI("ws://localhost:" + getHttpPort() +
"/graphql");
ClientUpgradeRequest request = new ClientUpgradeRequest();
+ request.setHeader("Authorization",
basicAuthHeader(BASIC_AUTH_USER_NAME, BASIC_AUTH_PASSWORD));
Future<Session> onConnected = client.connect(socket, echoUri,
request);
- RemoteEndpoint remote = onConnected.get().getRemote();
+ RemoteEndpoint remote = onConnected.get(10,
TimeUnit.SECONDS).getRemote();
LOGGER.info("Connected, initializing... ");
@@ -74,8 +92,7 @@ public class GraphQLWebSocketIT extends BaseGraphQLIT {
LOGGER.info("Waiting for socket to close...");
- CloseStatus status = socket.waitClose().get(10, TimeUnit.SECONDS);
- // Assert.assertEquals(1000, (int) status.getStatus()); TODO skip
for now
+ socket.waitClose().get(10, TimeUnit.SECONDS);
} finally {
client.stop();
@@ -83,6 +100,200 @@ public class GraphQLWebSocketIT extends BaseGraphQLIT {
}
}
+ /**
+ * A handshake carrying no credential is upgraded rather than refused,
because a browser cannot set
+ * request headers on a WebSocket handshake. The security property moved
rather than weakened: the
+ * socket that results can do nothing at all until it authenticates
through connection_init, which
+ * is what the following tests pin down.
+ */
+ @Test
+ public void testWebSocketUpgrade_withoutAuth_upgradesButCannotOperate()
throws Exception {
+ assertStartIsRefusedBeforeAuthentication(new ClientUpgradeRequest());
+ }
+
+ /** A public API key is not a subscription credential, on the handshake or
anywhere else. */
+ @Test
+ public void testWebSocketUpgrade_withPublicApiKeyOnly_cannotOperate()
throws Exception {
+ ClientUpgradeRequest request = new ClientUpgradeRequest();
+ request.setHeader("X-Unomi-Api-Key", testPublicKeyValue);
+ assertStartIsRefusedBeforeAuthentication(request);
+ }
+
+ /** connection_init carrying a valid credential is how a browser client
authenticates. */
+ @Test
+ public void
testWebSocketConnectionInit_withValidCredentials_authenticatesSocket() throws
Exception {
+ WebSocketClient client = new WebSocketClient();
+ Socket socket = new Socket();
+ try {
+ client.start();
+ Future<Session> onConnected = client.connect(socket,
graphqlWebSocketUri(), new ClientUpgradeRequest());
+ RemoteEndpoint remote = onConnected.get(10,
TimeUnit.SECONDS).getRemote();
+
+
remote.sendString(initWithCredentials(basicAuthHeader(TEST_TENANT_ID,
testPrivateKeyValue)));
+ String initResp = socket.waitMessage().get(10, TimeUnit.SECONDS);
+
Assert.assertEquals(resourceAsString("graphql/socket/in/ack.json"), initResp);
+
+
remote.sendString(resourceAsString("graphql/socket/out/term.json"));
+ socket.waitClose().get(10, TimeUnit.SECONDS);
+ } finally {
+ client.stop();
+ }
+ }
+
+ /** A wrong credential in connection_init must not authenticate the
socket. */
+ @Test
+ public void testWebSocketConnectionInit_withBadCredentials_isRefused()
throws Exception {
+ WebSocketClient client = new WebSocketClient();
+ Socket socket = new Socket();
+ try {
+ client.start();
+ Future<Session> onConnected = client.connect(socket,
graphqlWebSocketUri(), new ClientUpgradeRequest());
+ RemoteEndpoint remote = onConnected.get(10,
TimeUnit.SECONDS).getRemote();
+
+ // Subscribe for the server's connection_error before triggering
it: the client harness
+ // blocks in onWebSocketText until a listener exists, which would
otherwise stall the close.
+ Future<String> refusal = socket.waitMessage();
+
remote.sendString(resourceAsString("graphql/socket/out/init-bad-credentials.json"));
+
+ // Refused: the client is told, then the socket is closed rather
than acknowledged.
+ refusal.get(10, TimeUnit.SECONDS);
+ socket.waitClose().get(10, TimeUnit.SECONDS);
+ } finally {
+ client.stop();
+ }
+ }
+
+ /**
+ * The core property of the unauthenticated-upgrade path: an operation
sent before authenticating is
+ * refused and the socket is closed. Without this, opening the handshake
would be a regression.
+ */
+ private void assertStartIsRefusedBeforeAuthentication(ClientUpgradeRequest
request) throws Exception {
+ WebSocketClient client = new WebSocketClient();
+ Socket socket = new Socket();
+ try {
+ client.start();
+ Future<Session> onConnected = client.connect(socket,
graphqlWebSocketUri(), request);
+ RemoteEndpoint remote = onConnected.get(10,
TimeUnit.SECONDS).getRemote();
+
+ // Subscribe for the server's refusal message before triggering
it: the client harness
+ // blocks in onWebSocketText until a listener exists, which would
otherwise stall the close.
+ Future<String> refusal = socket.waitMessage();
+
remote.sendString(resourceAsString("graphql/socket/out/start.json"));
+
+ refusal.get(10, TimeUnit.SECONDS);
+ socket.waitClose().get(10, TimeUnit.SECONDS);
+ } finally {
+ client.stop();
+ }
+ }
+
+ private URI graphqlWebSocketUri() throws Exception {
+ return new URI("ws://localhost:" + getHttpPort() + "/graphql");
+ }
+
+ private String initWithCredentials(final String authorizationValue) {
+ return
resourceAsString("graphql/socket/out/init-with-credentials.json")
+ .replace("__AUTHORIZATION__", authorizationValue);
+ }
+
+ @Test
+ public void testWebSocketUpgrade_withWrongJaasPassword_returns401() throws
Exception {
+ ClientUpgradeRequest request = new ClientUpgradeRequest();
+ request.setHeader("Authorization",
basicAuthHeader(BASIC_AUTH_USER_NAME, "definitely-not-the-password"));
+ assertWebSocketUpgradeRejected(request);
+ }
+
+ @Test
+ public void testWebSocketUpgrade_withMalformedBasic_returns401() throws
Exception {
+ ClientUpgradeRequest request = new ClientUpgradeRequest();
+ request.setHeader("Authorization", "Basic !!!");
+ assertWebSocketUpgradeRejected(request);
+ }
+
+ @Test
+ public void testWebSocketUpgrade_withPrivateKey_succeeds() throws
Exception {
+ WebSocketClient client = new WebSocketClient();
+ Socket socket = new Socket();
+ try {
+ client.start();
+ URI echoUri = new URI("ws://localhost:" + getHttpPort() +
"/graphql");
+ ClientUpgradeRequest request = new ClientUpgradeRequest();
+ request.setHeader("Authorization", basicAuthHeader(TEST_TENANT_ID,
testPrivateKeyValue));
+
+ Future<Session> onConnected = client.connect(socket, echoUri,
request);
+ RemoteEndpoint remote = onConnected.get(10,
TimeUnit.SECONDS).getRemote();
+
+
remote.sendString(resourceAsString("graphql/socket/out/init.json"));
+ String initResp = socket.waitMessage().get(10, TimeUnit.SECONDS);
+
Assert.assertEquals(resourceAsString("graphql/socket/in/ack.json"), initResp);
+
+
remote.sendString(resourceAsString("graphql/socket/out/term.json"));
+ socket.waitClose().get(10, TimeUnit.SECONDS);
+ } finally {
+ client.stop();
+ }
+ }
+
+ /**
+ * Exercises GQL_START with the subject/context captured at upgrade time.
A successful
+ * subscription setup leaves the socket open (no error close); stop +
terminate then clean up.
+ */
+ @Test
+ public void testWebSocketSubscriptionStart_withPrivateKey_acceptsStart()
throws Exception {
+ WebSocketClient client = new WebSocketClient();
+ Socket socket = new Socket();
+ try {
+ client.start();
+ URI echoUri = new URI("ws://localhost:" + getHttpPort() +
"/graphql");
+ ClientUpgradeRequest request = new ClientUpgradeRequest();
+ request.setHeader("Authorization", basicAuthHeader(TEST_TENANT_ID,
testPrivateKeyValue));
+
+ Future<Session> onConnected = client.connect(socket, echoUri,
request);
+ RemoteEndpoint remote = onConnected.get(10,
TimeUnit.SECONDS).getRemote();
+ Future<CloseStatus> closeFuture = socket.waitClose();
+
+
remote.sendString(resourceAsString("graphql/socket/out/init.json"));
+ Assert.assertEquals(resourceAsString("graphql/socket/in/ack.json"),
+ socket.waitMessage().get(10, TimeUnit.SECONDS));
+
+
remote.sendString(resourceAsString("graphql/socket/out/start.json"));
+ // Successful subscribe() registers a publisher and does not emit
until events arrive.
+ // Give the server a moment; an auth/context failure would close
the socket with an error.
+ Thread.sleep(500);
+ Assert.assertFalse("Subscription start should not close the
socket", closeFuture.isDone());
+
+
remote.sendString(resourceAsString("graphql/socket/out/stop.json"));
+
remote.sendString(resourceAsString("graphql/socket/out/term.json"));
+ closeFuture.get(10, TimeUnit.SECONDS);
+ } finally {
+ client.stop();
+ }
+ }
+
+ private void assertWebSocketUpgradeRejected(ClientUpgradeRequest request)
throws Exception {
+ WebSocketClient client = new WebSocketClient();
+ Socket socket = new Socket();
+ try {
+ client.start();
+ URI echoUri = new URI("ws://localhost:" + getHttpPort() +
"/graphql");
+ Future<Session> onConnected = client.connect(socket, echoUri,
request);
+ try {
+ onConnected.get(10, TimeUnit.SECONDS);
+ Assert.fail("Unauthenticated GraphQL WebSocket upgrade should
be rejected");
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ Assert.assertTrue("Expected UpgradeException, got: " + cause,
cause instanceof UpgradeException);
+ Assert.assertEquals(401, ((UpgradeException)
cause).getResponseStatusCode());
+ }
+ } finally {
+ client.stop();
+ }
+ }
+
+ private static String basicAuthHeader(String user, String password) {
+ return "Basic " + Base64.getEncoder().encodeToString((user + ":" +
password).getBytes(StandardCharsets.UTF_8));
+ }
+
private class Socket extends WebSocketAdapter {
private Flowable<String> publisher;
diff --git
a/itests/src/test/resources/graphql/socket/out/init-bad-credentials.json
b/itests/src/test/resources/graphql/socket/out/init-bad-credentials.json
new file mode 100644
index 000000000..01af07218
--- /dev/null
+++ b/itests/src/test/resources/graphql/socket/out/init-bad-credentials.json
@@ -0,0 +1,6 @@
+{
+ "type": "connection_init",
+ "payload": {
+ "Authorization": "Basic bm9ib2R5Ondyb25n"
+ }
+}
diff --git
a/itests/src/test/resources/graphql/socket/out/init-with-credentials.json
b/itests/src/test/resources/graphql/socket/out/init-with-credentials.json
new file mode 100644
index 000000000..67e2565b2
--- /dev/null
+++ b/itests/src/test/resources/graphql/socket/out/init-with-credentials.json
@@ -0,0 +1,6 @@
+{
+ "type": "connection_init",
+ "payload": {
+ "Authorization": "__AUTHORIZATION__"
+ }
+}
