From f0fdcf7fd6f61c861ac9a620f3794e012b589b38 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 18 Sep 2026 12:56:58 +0300 Subject: [PATCH 1/4] Close websocket sessions reliably when the upstream is missing ServerEndPoint dereferenced session_upstream after waiting for it, relying on the resulting NullPointerException to close the client. With the long NPE messages of Java 14+ the close never happened: CloseReason rejects reason phrases over 123 UTF-8 bytes, the IllegalArgumentException was swallowed, and the client connection was left open. - wait for the upstream in upstream() and fail with an explicit "upstream not connected" IOException - build every CloseReason through closeReason(), which truncates the phrase to the 123-byte protocol limit - close the upstream null-safely in one place and make the field volatile (java/dereferenced-value-may-be-null) - SeparatedValuesFile: open the reader in try-with-resources (java/input-resource-leak) --- .../openig/text/SeparatedValuesFile.java | 9 +- .../openig/text/SeparatedValuesReader.java | 5 +- .../openig/websocket/ServerEndPoint.java | 110 ++++++++++-------- .../openig/websocket/ServerEndPointTest.java | 94 +++++++++++++++ 4 files changed, 164 insertions(+), 54 deletions(-) create mode 100644 openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java diff --git a/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesFile.java b/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesFile.java index 67ac49c09..da19e3728 100644 --- a/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesFile.java +++ b/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesFile.java @@ -13,6 +13,7 @@ * * Copyright 2010-2011 ApexIdentity Inc. * Portions Copyright 2011-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.forgerock.openig.text; @@ -148,11 +149,9 @@ public List getFields() { */ public Map getRecord(String key, String value) throws IOException { Map map = null; - SeparatedValuesReader reader = new SeparatedValuesReader( + try (SeparatedValuesReader reader = new SeparatedValuesReader( new InputStreamReader(new FileInputStream(file), charset), - separator - ); - try { + separator)) { List fields = this.fields; if (header) { // first line in the file is the field header @@ -181,8 +180,6 @@ public Map getRecord(String key, String value) throws IOExceptio } } } - } finally { - reader.close(); } return map; } diff --git a/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesReader.java b/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesReader.java index 89af4befc..79b8662ec 100644 --- a/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesReader.java +++ b/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesReader.java @@ -13,12 +13,14 @@ * * Copyright 2010-2011 ApexIdentity Inc. * Portions Copyright 2011-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.forgerock.openig.text; import static org.forgerock.util.Utils.closeSilently; +import java.io.Closeable; import java.io.IOException; import java.io.Reader; import java.util.ArrayList; @@ -27,7 +29,7 @@ /** * Reads records with delimiter-separated values from a character stream. */ -public class SeparatedValuesReader { +public class SeparatedValuesReader implements Closeable { private static final int CR = '\r'; private static final int LF = '\n'; @@ -113,6 +115,7 @@ public List next() throws IOException { * reader has been closed, further {@code next()} invocations will throw an * {@code IOException}. Closing a previously closed reader has no effect. */ + @Override public void close() { closeSilently(input); } diff --git a/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java b/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java index 3395bfb62..17f497d38 100644 --- a/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java +++ b/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java @@ -11,7 +11,7 @@ * Header, with the fields enclosed by brackets [] replaced by your own identifying * information: "Portions copyright [year] [name of copyright owner]". * - * Copyright 2022-2025 3A Systems LLC. + * Copyright 2022-2026 3A Systems LLC. */ package org.openidentityplatform.openig.websocket; @@ -35,6 +35,7 @@ import jakarta.websocket.server.ServerEndpoint; import java.io.IOException; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; import java.util.Map.Entry; @@ -45,7 +46,52 @@ public class ServerEndPoint { static Logger logger=LoggerFactory.getLogger(ServerEndPoint.class); Principal principal=null; - Session session_upstream=null; + volatile Session session_upstream=null; + + /** A close frame reason phrase is limited to 123 UTF-8 bytes; {@link CloseReason} rejects longer ones. */ + static final int MAX_REASON_PHRASE_BYTES = 123; + + /** Builds a {@link CloseReason}, truncating the phrase so that it never exceeds the protocol limit. */ + static CloseReason closeReason(CloseReason.CloseCode code, String phrase) { + String reason = phrase.length() > MAX_REASON_PHRASE_BYTES ? phrase.substring(0, MAX_REASON_PHRASE_BYTES) : phrase; + while (reason.getBytes(StandardCharsets.UTF_8).length > MAX_REASON_PHRASE_BYTES) { + reason = reason.substring(0, reason.length() - 1); + } + return new CloseReason(code, reason); + } + + /** Waits (up to ~5 s) for the upstream session to be connected and returns it. */ + private Session upstream() throws IOException, InterruptedException { + int ct=0; + while ((ct++<5000)&&(session_upstream==null || !session_upstream.isOpen())) { + Thread.sleep(1); + } + final Session upstream=session_upstream; + if (upstream==null || !upstream.isOpen()) { + throw new IOException("upstream not connected"); + } + return upstream; + } + + /** Closes the upstream session (if any) with the given reason and forgets it. */ + private void closeUpstream(CloseReason reason) { + final Session upstream=session_upstream; + session_upstream=null; + if (upstream!=null) { + try { + upstream.close(reason); + } catch (Throwable e1) {} + } + } + + /** Closes both the client and the upstream sessions abnormally with the given reason. */ + private void closeAbnormally(Session session_client, String phrase) { + final CloseReason reason=closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, phrase); + try { + session_client.close(reason); + } catch (Throwable e1) {} + closeUpstream(reason); + } public void authorize() throws Exception { if (principal==null || principal.authorize().getCode()!=101) { @@ -101,11 +147,12 @@ public void onMessage(String message) { authorize(); session_client.getBasicRemote().sendText(message); } catch (Throwable e) { + final CloseReason reason=closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString()); try { - session_client.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString())); + session_client.close(reason); } catch (Throwable e1) {} try { - session_upstream.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString())); + session_upstream.close(reason); } catch (Throwable e1) {} } @@ -121,11 +168,12 @@ public void onMessage(ByteBuffer message) { authorize(); session_client.getBasicRemote().sendBinary(message); } catch (Throwable e) { + final CloseReason reason=closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString()); try { - session_client.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString())); + session_client.close(reason); } catch (Throwable e1) {} try { - session_upstream.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString())); + session_upstream.close(reason); } catch (Throwable e1) {} } } @@ -146,7 +194,7 @@ public void onClose(Session session_upstream, CloseReason reason) { } catch (Exception e) { logger.error("{}: {}",principal.request.getUri().asURI(),e.toString()); try { - session_client.close(new CloseReason(CloseReason.CloseCodes.TRY_AGAIN_LATER,"upstream down: "+e.toString())); + session_client.close(closeReason(CloseReason.CloseCodes.TRY_AGAIN_LATER,"upstream down: "+e.toString())); } catch (IOException e1) {} } } @@ -155,12 +203,7 @@ public void onClose(Session session_upstream, CloseReason reason) { public void end(Session session_client,CloseReason reason) throws IOException { logger.debug("client close {} {}: {}",session_client.getRequestURI(),session_client.getId(),reason); - try { - session_upstream.close(reason); - } catch (Throwable ioe) {} - finally { - session_upstream=null; - } + closeUpstream(reason); } @OnError @@ -168,12 +211,7 @@ public void onError(Session session_client,Throwable t) throws Throwable { try { session_client.close(); } catch (Throwable e1) {} - try { - session_upstream.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client error: "+t.toString())); - } catch (Throwable e1) {} - finally { - session_upstream=null; - } + closeUpstream(closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client error: "+t.toString())); } @OnMessage @@ -182,22 +220,11 @@ public void echoTextMessage(Session session_client,String msg, boolean last) { logger.trace("->{}: {}",session_client.getRequestURI(),msg); } try { - int ct=0; - while ((ct++<5000)&&(session_upstream==null || !session_upstream.isOpen())) { - Thread.sleep(1); - } + final Session upstream=upstream(); authorize(); - session_upstream.getBasicRemote().sendText(msg, last); + upstream.getBasicRemote().sendText(msg, last); } catch (Throwable e) { - try { - session_client.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client message error: "+e.toString())); - } catch (Throwable e1) {} - try { - session_upstream.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client message error: "+e.toString())); - } catch (Throwable e1) {} - finally { - session_upstream=null; - } + closeAbnormally(session_client, "client message error: "+e.toString()); } } @@ -207,22 +234,11 @@ public void echoBinaryMessage(Session session_client,ByteBuffer bb,boolean last) logger.trace("->{}: {}",session_client.getRequestURI(),bb.capacity()); } try { - int ct=0; - while ((ct++<5000)&&(session_upstream==null || !session_upstream.isOpen())) { - Thread.sleep(1); - } + final Session upstream=upstream(); authorize(); - session_upstream.getBasicRemote().sendBinary(bb, last); + upstream.getBasicRemote().sendBinary(bb, last); } catch (Throwable e) { - try { - session_client.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client message error: "+e.toString())); - } catch (Throwable e1) {} - try { - session_upstream.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client message error: "+e.toString())); - } catch (Throwable e1) {} - finally { - session_upstream=null; - } + closeAbnormally(session_client, "client message error: "+e.toString()); } } } diff --git a/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java b/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java new file mode 100644 index 000000000..5652efbc9 --- /dev/null +++ b/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java @@ -0,0 +1,94 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ + +package org.openidentityplatform.openig.websocket; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; + +import org.forgerock.http.protocol.Status; +import org.mockito.ArgumentCaptor; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import jakarta.websocket.CloseReason; +import jakarta.websocket.Session; + +@SuppressWarnings("javadoc") +public class ServerEndPointTest { + + private ServerEndPoint endpoint; + private Session client; + + @BeforeMethod + public void setUp() throws Exception { + endpoint = new ServerEndPoint(); + endpoint.principal = mock(Principal.class); + when(endpoint.principal.authorize()).thenReturn(Status.SWITCHING_PROTOCOLS); + client = mock(Session.class); + // No upstream session: the endpoint waits for it (up to ~5 s) and then must give up cleanly + endpoint.session_upstream = null; + } + + @Test + public void shouldCloseClientWithClearReasonWhenUpstreamIsNotConnectedOnTextMessage() throws Exception { + endpoint.echoTextMessage(client, "hello", true); + + CloseReason reason = closeReasonOf(client); + assertThat(reason.getCloseCode()).isEqualTo(CloseReason.CloseCodes.CLOSED_ABNORMALLY); + assertThat(reason.getReasonPhrase()) + .contains("upstream not connected") + .doesNotContain("NullPointerException"); + } + + @Test + public void shouldCloseClientWithClearReasonWhenUpstreamIsNotConnectedOnBinaryMessage() throws Exception { + endpoint.echoBinaryMessage(client, ByteBuffer.wrap(new byte[] { 1, 2, 3 }), true); + + CloseReason reason = closeReasonOf(client); + assertThat(reason.getCloseCode()).isEqualTo(CloseReason.CloseCodes.CLOSED_ABNORMALLY); + assertThat(reason.getReasonPhrase()) + .contains("upstream not connected") + .doesNotContain("NullPointerException"); + } + + @Test + public void shouldTruncateCloseReasonPhraseToProtocolLimit() { + // ASCII: cut to 123 bytes + String ascii = "x".repeat(500); + assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, ascii).getReasonPhrase()) + .isEqualTo("x".repeat(123)); + // Multi-byte: never split a character, never exceed 123 UTF-8 bytes + String cyrillic = "\u0436".repeat(200); + String phrase = ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, cyrillic).getReasonPhrase(); + assertThat(phrase.getBytes(StandardCharsets.UTF_8)).hasSize(122); + assertThat(phrase).isEqualTo("\u0436".repeat(61)); + // Short phrases are left untouched + assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "short").getReasonPhrase()) + .isEqualTo("short"); + } + + private static CloseReason closeReasonOf(Session session) throws Exception { + ArgumentCaptor reason = ArgumentCaptor.forClass(CloseReason.class); + verify(session).close(reason.capture()); + return reason.getValue(); + } +} From 1a122ed1a06acb2bec98e59b93b2cae7d452c6f4 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 18 Sep 2026 13:37:36 +0300 Subject: [PATCH 2/4] Add @Override to the upstream message handlers in ServerEndPoint Closes the java/missing-override-annotation notes #23 and #24. --- .../openidentityplatform/openig/websocket/ServerEndPoint.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java b/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java index 17f497d38..ba1943ef8 100644 --- a/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java +++ b/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java @@ -139,6 +139,7 @@ public void onOpen(Session session_upstream, EndpointConfig config) { session_upstream.setMaxTextMessageBufferSize(session_client.getMaxTextMessageBufferSize()); session_upstream.setMaxIdleTimeout(session_client.getMaxIdleTimeout()); session_upstream.addMessageHandler(new MessageHandler.Whole() { + @Override public void onMessage(String message) { try { if (logger.isTraceEnabled()) { @@ -160,6 +161,7 @@ public void onMessage(String message) { }); session_upstream.addMessageHandler(new MessageHandler.Whole() { + @Override public void onMessage(ByteBuffer message) { try { if (logger.isTraceEnabled()) { From f3c8770138080e2074f9622c69dc6075102609fd Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Thu, 22 Oct 2026 15:05:35 +0300 Subject: [PATCH 3/4] Address review on websocket close handling - closeReason(): never end the phrase in a lone high surrogate; a dangling half is encoded as a single '?' byte and slipped past the length check - upstream(): read the volatile field once per iteration instead of twice in the wait loop - SeparatedValuesFile: open the FileInputStream as a resource of its own so it is closed even if the InputStreamReader constructor throws - tests: pin the surrogate cut, the truncation of a long upstream error in closeAbnormally(), and the close of a stale (not open) upstream --- .../openig/text/SeparatedValuesFile.java | 7 +-- .../openig/websocket/ServerEndPoint.java | 14 ++++-- .../openig/websocket/ServerEndPointTest.java | 44 +++++++++++++++++++ 3 files changed, 58 insertions(+), 7 deletions(-) diff --git a/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesFile.java b/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesFile.java index da19e3728..3a2bf40c1 100644 --- a/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesFile.java +++ b/openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesFile.java @@ -149,9 +149,10 @@ public List getFields() { */ public Map getRecord(String key, String value) throws IOException { Map map = null; - try (SeparatedValuesReader reader = new SeparatedValuesReader( - new InputStreamReader(new FileInputStream(file), charset), - separator)) { + try (FileInputStream stream = new FileInputStream(file); + SeparatedValuesReader reader = new SeparatedValuesReader( + new InputStreamReader(stream, charset), + separator)) { List fields = this.fields; if (header) { // first line in the file is the field header diff --git a/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java b/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java index ba1943ef8..a274c8a10 100644 --- a/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java +++ b/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java @@ -51,10 +51,15 @@ public class ServerEndPoint { /** A close frame reason phrase is limited to 123 UTF-8 bytes; {@link CloseReason} rejects longer ones. */ static final int MAX_REASON_PHRASE_BYTES = 123; - /** Builds a {@link CloseReason}, truncating the phrase so that it never exceeds the protocol limit. */ + /** + * Builds a {@link CloseReason}, truncating the phrase so that it never exceeds the protocol limit + * and never ends in the first half of a surrogate pair. + */ static CloseReason closeReason(CloseReason.CloseCode code, String phrase) { String reason = phrase.length() > MAX_REASON_PHRASE_BYTES ? phrase.substring(0, MAX_REASON_PHRASE_BYTES) : phrase; - while (reason.getBytes(StandardCharsets.UTF_8).length > MAX_REASON_PHRASE_BYTES) { + // a lone high surrogate is encoded as a single '?' byte, so the length check alone does not catch it + while (reason.getBytes(StandardCharsets.UTF_8).length > MAX_REASON_PHRASE_BYTES + || (!reason.isEmpty() && Character.isHighSurrogate(reason.charAt(reason.length() - 1)))) { reason = reason.substring(0, reason.length() - 1); } return new CloseReason(code, reason); @@ -63,10 +68,11 @@ static CloseReason closeReason(CloseReason.CloseCode code, String phrase) { /** Waits (up to ~5 s) for the upstream session to be connected and returns it. */ private Session upstream() throws IOException, InterruptedException { int ct=0; - while ((ct++<5000)&&(session_upstream==null || !session_upstream.isOpen())) { + Session upstream=session_upstream; + while ((ct++<5000)&&(upstream==null || !upstream.isOpen())) { Thread.sleep(1); + upstream=session_upstream; } - final Session upstream=session_upstream; if (upstream==null || !upstream.isOpen()) { throw new IOException("upstream not connected"); } diff --git a/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java b/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java index 5652efbc9..8841374d0 100644 --- a/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java +++ b/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java @@ -17,10 +17,13 @@ package org.openidentityplatform.openig.websocket; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.IOException; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; @@ -30,6 +33,7 @@ import org.testng.annotations.Test; import jakarta.websocket.CloseReason; +import jakarta.websocket.RemoteEndpoint; import jakarta.websocket.Session; @SuppressWarnings("javadoc") @@ -70,6 +74,40 @@ public void shouldCloseClientWithClearReasonWhenUpstreamIsNotConnectedOnBinaryMe .doesNotContain("NullPointerException"); } + @Test + public void shouldTruncateLongUpstreamErrorWhenClosingClient() throws Exception { + // Connected upstream that fails with a message longer than a close frame can carry + Session upstream = mock(Session.class); + RemoteEndpoint.Basic remote = mock(RemoteEndpoint.Basic.class); + when(upstream.isOpen()).thenReturn(true); + when(upstream.getBasicRemote()).thenReturn(remote); + doThrow(new IOException("x".repeat(200))).when(remote).sendText("hello", true); + endpoint.session_upstream = upstream; + + endpoint.echoTextMessage(client, "hello", true); + + CloseReason reason = closeReasonOf(client); + assertThat(reason.getCloseCode()).isEqualTo(CloseReason.CloseCodes.CLOSED_ABNORMALLY); + assertThat(reason.getReasonPhrase()).startsWith("client message error: java.io.IOException: x"); + assertThat(reason.getReasonPhrase().getBytes(StandardCharsets.UTF_8)).hasSize(123); + verify(upstream).close(reason); + assertThat(endpoint.session_upstream).isNull(); + } + + @Test + public void shouldCloseStaleUpstreamWhenItIsNotOpen() throws Exception { + // Upstream session is set but already closed: the endpoint waits for it (up to ~5 s), then drops it + Session upstream = mock(Session.class); + when(upstream.isOpen()).thenReturn(false); + endpoint.session_upstream = upstream; + + endpoint.echoTextMessage(client, "hello", true); + + assertThat(closeReasonOf(client).getReasonPhrase()).contains("upstream not connected"); + verify(upstream).close(any(CloseReason.class)); + assertThat(endpoint.session_upstream).isNull(); + } + @Test public void shouldTruncateCloseReasonPhraseToProtocolLimit() { // ASCII: cut to 123 bytes @@ -81,6 +119,12 @@ public void shouldTruncateCloseReasonPhraseToProtocolLimit() { String phrase = ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, cyrillic).getReasonPhrase(); assertThat(phrase.getBytes(StandardCharsets.UTF_8)).hasSize(122); assertThat(phrase).isEqualTo("\u0436".repeat(61)); + // Surrogate pairs: never leave the high half of a pair at the end of the phrase + String smile = "\uD83D\uDE00"; + assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "x" + smile.repeat(100)).getReasonPhrase()) + .isEqualTo("x" + smile.repeat(30)); + assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "a".repeat(122) + smile.repeat(5)).getReasonPhrase()) + .isEqualTo("a".repeat(122)); // Short phrases are left untouched assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "short").getReasonPhrase()) .isEqualTo("short"); From d1a4ade7d0cdb8d4caa13a549d63ce05668bb64c Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Thu, 22 Oct 2026 21:47:49 +0300 Subject: [PATCH 4/4] Address second review round on websocket close handling - closeReason(): mark a cut phrase with U+2026 after at most 120 bytes, as Tomcat's WsSession does, so a truncated reason reads the same on every container - echo handlers: call authorize() before waiting for the upstream, so a revoked client is closed with "access revoked" without the ~5 s wait - onError(): close the upstream before the client; on Tomcat close() runs @OnClose first and the upstream never got the client error - tests: SeparatedValuesFileTest for getRecord() (FileAttributesFilterTest mocks SeparatedValuesFile), pins for the authorize() order, the onError() order and the empty-phrase guard in closeReason() --- .../openig/websocket/ServerEndPoint.java | 24 +++++-- .../openig/text/SeparatedValuesFileTest.java | 47 ++++++++++++++ .../openig/websocket/ServerEndPointTest.java | 65 ++++++++++++++++--- 3 files changed, 120 insertions(+), 16 deletions(-) create mode 100644 openig-core/src/test/java/org/forgerock/openig/text/SeparatedValuesFileTest.java diff --git a/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java b/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java index a274c8a10..63d7ac855 100644 --- a/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java +++ b/openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java @@ -53,16 +53,25 @@ public class ServerEndPoint { /** * Builds a {@link CloseReason}, truncating the phrase so that it never exceeds the protocol limit - * and never ends in the first half of a surrogate pair. + * and never ends in the first half of a surrogate pair. A truncated phrase ends with U+2026. */ static CloseReason closeReason(CloseReason.CloseCode code, String phrase) { - String reason = phrase.length() > MAX_REASON_PHRASE_BYTES ? phrase.substring(0, MAX_REASON_PHRASE_BYTES) : phrase; + if (phrase.getBytes(StandardCharsets.UTF_8).length <= MAX_REASON_PHRASE_BYTES) { + return new CloseReason(code, truncate(phrase, MAX_REASON_PHRASE_BYTES)); + } + // mark the cut as Tomcat does: at most 120 bytes, then U+2026 (3 bytes in UTF-8) + return new CloseReason(code, truncate(phrase, MAX_REASON_PHRASE_BYTES - 3) + "\u2026"); + } + + /** Cuts the phrase to at most {@code maxBytes} UTF-8 bytes without splitting a character. */ + private static String truncate(String phrase, int maxBytes) { + String reason = phrase.length() > maxBytes ? phrase.substring(0, maxBytes) : phrase; // a lone high surrogate is encoded as a single '?' byte, so the length check alone does not catch it - while (reason.getBytes(StandardCharsets.UTF_8).length > MAX_REASON_PHRASE_BYTES + while (reason.getBytes(StandardCharsets.UTF_8).length > maxBytes || (!reason.isEmpty() && Character.isHighSurrogate(reason.charAt(reason.length() - 1)))) { reason = reason.substring(0, reason.length() - 1); } - return new CloseReason(code, reason); + return reason; } /** Waits (up to ~5 s) for the upstream session to be connected and returns it. */ @@ -216,10 +225,11 @@ public void end(Session session_client,CloseReason reason) throws IOException { @OnError public void onError(Session session_client,Throwable t) throws Throwable { + // upstream first: Tomcat runs @OnClose inside close(), and end() would close the upstream normally + closeUpstream(closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client error: "+t.toString())); try { session_client.close(); } catch (Throwable e1) {} - closeUpstream(closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client error: "+t.toString())); } @OnMessage @@ -228,8 +238,8 @@ public void echoTextMessage(Session session_client,String msg, boolean last) { logger.trace("->{}: {}",session_client.getRequestURI(),msg); } try { - final Session upstream=upstream(); authorize(); + final Session upstream=upstream(); upstream.getBasicRemote().sendText(msg, last); } catch (Throwable e) { closeAbnormally(session_client, "client message error: "+e.toString()); @@ -242,8 +252,8 @@ public void echoBinaryMessage(Session session_client,ByteBuffer bb,boolean last) logger.trace("->{}: {}",session_client.getRequestURI(),bb.capacity()); } try { - final Session upstream=upstream(); authorize(); + final Session upstream=upstream(); upstream.getBasicRemote().sendBinary(bb, last); } catch (Throwable e) { closeAbnormally(session_client, "client message error: "+e.toString()); diff --git a/openig-core/src/test/java/org/forgerock/openig/text/SeparatedValuesFileTest.java b/openig-core/src/test/java/org/forgerock/openig/text/SeparatedValuesFileTest.java new file mode 100644 index 000000000..c729d4f7d --- /dev/null +++ b/openig-core/src/test/java/org/forgerock/openig/text/SeparatedValuesFileTest.java @@ -0,0 +1,47 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ + +package org.forgerock.openig.text; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.assertj.core.api.Assertions.assertThat; + +import java.nio.file.Files; +import java.nio.file.Path; + +import org.testng.annotations.Test; + +@SuppressWarnings("javadoc") +public class SeparatedValuesFileTest { + + @Test + public void shouldReadRecordFromFile() throws Exception { + Path csv = Files.createTempFile("users", ".csv"); + try { + Files.write(csv, "username,password\njoe,secret\nbob,hunter2\n".getBytes(UTF_8)); + + SeparatedValuesFile file = new SeparatedValuesFile(csv.toFile()); + + assertThat(file.getRecord("username", "bob")) + .containsEntry("username", "bob") + .containsEntry("password", "hunter2"); + assertThat(file.getRecord("username", "nobody")).isNull(); + assertThat(file.getRecord("missing", "joe")).isNull(); + } finally { + Files.delete(csv); + } + } +} diff --git a/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java b/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java index 8841374d0..54462c95a 100644 --- a/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java +++ b/openig-core/src/test/java/org/openidentityplatform/openig/websocket/ServerEndPointTest.java @@ -18,6 +18,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; @@ -88,12 +89,52 @@ public void shouldTruncateLongUpstreamErrorWhenClosingClient() throws Exception CloseReason reason = closeReasonOf(client); assertThat(reason.getCloseCode()).isEqualTo(CloseReason.CloseCodes.CLOSED_ABNORMALLY); - assertThat(reason.getReasonPhrase()).startsWith("client message error: java.io.IOException: x"); + assertThat(reason.getReasonPhrase()) + .startsWith("client message error: java.io.IOException: x") + .endsWith("\u2026"); assertThat(reason.getReasonPhrase().getBytes(StandardCharsets.UTF_8)).hasSize(123); verify(upstream).close(reason); assertThat(endpoint.session_upstream).isNull(); } + @Test + public void shouldCloseRevokedClientWithoutWaitingForUpstreamOnTextMessage() throws Exception { + // Access is checked before the endpoint waits for the upstream + when(endpoint.principal.authorize()).thenReturn(Status.FORBIDDEN); + + endpoint.echoTextMessage(client, "hello", true); + + assertThat(closeReasonOf(client).getReasonPhrase()).isEqualTo("client message error: access revoked"); + } + + @Test + public void shouldCloseRevokedClientWithoutWaitingForUpstreamOnBinaryMessage() throws Exception { + when(endpoint.principal.authorize()).thenReturn(Status.FORBIDDEN); + + endpoint.echoBinaryMessage(client, ByteBuffer.wrap(new byte[] { 1, 2, 3 }), true); + + assertThat(closeReasonOf(client).getReasonPhrase()).isEqualTo("client message error: access revoked"); + } + + @Test + public void shouldCloseUpstreamWithClientErrorOnError() throws Throwable { + Session upstream = mock(Session.class); + endpoint.session_upstream = upstream; + // Tomcat runs @OnClose on the calling thread, from inside close() + doAnswer(invocation -> { + endpoint.end(client, new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "")); + return null; + }).when(client).close(); + + endpoint.onError(client, new IOException("broken pipe")); + + CloseReason reason = closeReasonOf(upstream); + assertThat(reason.getCloseCode()).isEqualTo(CloseReason.CloseCodes.CLOSED_ABNORMALLY); + assertThat(reason.getReasonPhrase()).isEqualTo("client error: java.io.IOException: broken pipe"); + verify(client).close(); + assertThat(endpoint.session_upstream).isNull(); + } + @Test public void shouldCloseStaleUpstreamWhenItIsNotOpen() throws Exception { // Upstream session is set but already closed: the endpoint waits for it (up to ~5 s), then drops it @@ -110,24 +151,30 @@ public void shouldCloseStaleUpstreamWhenItIsNotOpen() throws Exception { @Test public void shouldTruncateCloseReasonPhraseToProtocolLimit() { - // ASCII: cut to 123 bytes - String ascii = "x".repeat(500); - assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, ascii).getReasonPhrase()) + String ellipsis = "\u2026"; + // ASCII: up to 123 bytes untouched, longer cut to 120 bytes plus the 3-byte ellipsis + assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "x".repeat(123)).getReasonPhrase()) .isEqualTo("x".repeat(123)); + assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "x".repeat(500)).getReasonPhrase()) + .isEqualTo("x".repeat(120) + ellipsis); // Multi-byte: never split a character, never exceed 123 UTF-8 bytes String cyrillic = "\u0436".repeat(200); String phrase = ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, cyrillic).getReasonPhrase(); - assertThat(phrase.getBytes(StandardCharsets.UTF_8)).hasSize(122); - assertThat(phrase).isEqualTo("\u0436".repeat(61)); + assertThat(phrase.getBytes(StandardCharsets.UTF_8)).hasSize(123); + assertThat(phrase).isEqualTo("\u0436".repeat(60) + ellipsis); // Surrogate pairs: never leave the high half of a pair at the end of the phrase String smile = "\uD83D\uDE00"; assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "x" + smile.repeat(100)).getReasonPhrase()) - .isEqualTo("x" + smile.repeat(30)); - assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "a".repeat(122) + smile.repeat(5)).getReasonPhrase()) - .isEqualTo("a".repeat(122)); + .isEqualTo("x" + smile.repeat(29) + ellipsis); + assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "a".repeat(119) + smile.repeat(5)).getReasonPhrase()) + .isEqualTo("a".repeat(119) + ellipsis); + assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "\uD83D").getReasonPhrase()) + .isEmpty(); // Short phrases are left untouched assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "short").getReasonPhrase()) .isEqualTo("short"); + assertThat(ServerEndPoint.closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, "").getReasonPhrase()) + .isEmpty(); } private static CloseReason closeReasonOf(Session session) throws Exception {