From 5cbba7853b8e020a2fd8d45f13e88b136fca82fc Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Thu, 22 Oct 2026 00:50:34 +0200 Subject: [PATCH] fix: harden WHIP/WHEP session setup and teardown A resource the server names after the session timed out, or before its answer failed to read, is now deleted instead of left on the server. The failure is reported before that deletion runs. The bearer token no longer follows redirects or a Location that drops TLS. DELETE checks its status and reports failures, response bodies are size-limited, and the session timer is cancelled once it is settled. --- docs/guide/client-api.md | 4 +- .../dev/onvoid/webrtc/client/HttpSession.java | 195 +++++++++++++----- .../webrtc/client/HttpSessionTests.java | 109 ++++++++++ .../webrtc/client/TestHttpEndpoint.java | 46 ++++- 4 files changed, 293 insertions(+), 61 deletions(-) diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index 26995757..d7abe8cb 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -537,7 +537,9 @@ viewer.peer().onTrack(track -> System.out.println("Receiving " + track.kind())); The offer goes to the endpoint with the local candidates, as WHIP and WHEP servers expect, and the future completes once the peer is connected. If the endpoint refuses the offer, or the peer does not connect within the timeout (15 seconds by default), the future fails, and the session is deleted on the server again. `onTrack` hands over tracks that already arrived, so a handler added once the future completed misses none. -`session.close()` deletes the session on the server, which frees what the server holds for it, and closes the peer. The tracks a WHIP session published stay open. `session.peer()` is a regular peer, for stats and anything else. +`session.close()` deletes the session on the server, which frees what the server holds for it, and closes the peer. The tracks a WHIP session published stay open. `session.peer()` is a regular peer, for stats and anything else. A deletion the server refuses or that fails goes to the runtime's error handler. + +The bearer token goes to the endpoint, and along with the deletion to the session the endpoint names in its `Location` header. That session may be on another host, but an `https` endpoint cannot name a plain `http` one. Redirects are not followed, since they would take the token elsewhere: a redirected request fails with its status. | Builder method | What it sets | |---|---| diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java index c0843c36..974ff1ed 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java @@ -27,6 +27,8 @@ import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -42,6 +44,12 @@ */ abstract class HttpSession implements AutoCloseable { + /** The largest answer read; an SDP answer is a few kilobytes. */ + private static final int MAX_BODY = 1 << 20; + + /** How much of an error response goes into a message. */ + private static final int MAX_ERROR_BODY = 1024; + private final Peer peer; private final URI resource; @@ -88,17 +96,18 @@ public URI resource() { */ public CompletableFuture closeAsync() { if (closing.compareAndSet(false, true)) { - CompletableFuture.runAsync(() -> { - try { - if (resource != null) { - delete(resource, bearerToken, timeout); - } - } - catch (Exception e) { - rtc.reportError(new WebRtcException("Deleting the session " + resource + " failed", e)); - } - }, rtc.controlPool) - .thenCompose(ignored -> peer.closeAsync()) + CompletableFuture deleted; + + try { + deleted = CompletableFuture.runAsync(() -> discard(rtc, resource, bearerToken, timeout), + rtc.controlPool); + } + catch (RejectedExecutionException e) { + leftOnServer(rtc, resource); + deleted = CompletableFuture.completedFuture(null); + } + + deleted.thenCompose(ignored -> peer.closeAsync()) .whenComplete((ignored, error) -> closed.complete(null)); } @@ -122,6 +131,8 @@ public void close() { static CompletableFuture negotiate(WebRtc rtc, Peer peer, URI endpoint, String bearerToken, Duration timeout) { CompletableFuture result = new CompletableFuture<>(); + // The resource the server created. Whoever takes it out owns it: the + // session that connected, or whoever deletes it. AtomicReference resource = new AtomicReference<>(); AtomicBoolean finished = new AtomicBoolean(); @@ -131,50 +142,66 @@ static CompletableFuture negotiate(WebRtc rtc, Peer peer, URI endpoint, Str return; } - // Leaves nothing behind, on the server or here. - CompletableFuture.runAsync(() -> { - try { - URI created = resource.get(); + // Reported before the resource is deleted, which may take as long + // as the timeout again. + peer.close(); + result.completeExceptionally(PeerImpl.asWebRtcException(error, + "The session with " + endpoint + " could not be set up")); - if (created != null) { - delete(created, bearerToken, timeout); - } + URI created = resource.getAndSet(null); + + if (created != null) { + try { + CompletableFuture.runAsync(() -> discard(rtc, created, bearerToken, timeout), rtc.controlPool); } - catch (Exception e) { - // The failure that ended the session is the one to report. + catch (RejectedExecutionException e) { + leftOnServer(rtc, created); } - - peer.close(); - result.completeExceptionally(PeerImpl.asWebRtcException(error, - "The session with " + endpoint + " could not be set up")); - }, rtc.controlPool); + } }; + ScheduledFuture deadline = rtc.timer.schedule(() -> fail.accept(new WebRtcException( + "The session with " + endpoint + " was not connected within " + timeout)), + timeout.toMillis(), TimeUnit.MILLISECONDS); + peer.negotiator().createOffer(Gathering.COMPLETE) .thenApplyAsync(offer -> { - Response response = post(endpoint, offer.sdp, bearerToken, timeout); - resource.set(response.location); - return response.body; + try { + return post(endpoint, offer.sdp, bearerToken, timeout, resource); + } + finally { + if (finished.get()) { + // The session ended while the offer was out, so what + // the server created meanwhile has no one else to + // delete it. The answer then meets a closed peer. + discard(rtc, resource.getAndSet(null), bearerToken, timeout); + } + } }, rtc.controlPool) .thenCompose(answer -> peer.negotiator() .applyAnswer(new RTCSessionDescription(RTCSdpType.ANSWER, answer))) .thenCompose(applied -> peer.whenConnected()) .whenComplete((connected, error) -> { + deadline.cancel(false); + if (error != null) { fail.accept(error); } else if (finished.compareAndSet(false, true)) { - result.complete(resource.get()); + result.complete(resource.getAndSet(null)); } }); - rtc.timer.schedule(() -> fail.accept(new WebRtcException("The session with " + endpoint - + " was not connected within " + timeout)), timeout.toMillis(), TimeUnit.MILLISECONDS); - return result; } - private static Response post(URI endpoint, String sdp, String bearerToken, Duration timeout) { + /** + * Posts the offer and returns the answer. The resource the server names is + * set as soon as it is known, so that it is deleted even if reading the + * answer fails. + */ + private static String post(URI endpoint, String sdp, String bearerToken, Duration timeout, + AtomicReference resource) { HttpURLConnection connection = null; try { @@ -194,13 +221,12 @@ private static Response post(URI endpoint, String sdp, String bearerToken, Durat if (status != HttpURLConnection.HTTP_CREATED && status != HttpURLConnection.HTTP_OK) { throw new WebRtcException("The endpoint " + endpoint + " answered the offer with status " + status - + ": " + read(connection.getErrorStream())); + + ": " + read(connection.getErrorStream(), MAX_ERROR_BODY, true)); } - String location = connection.getHeaderField("Location"); + resource.set(resourceOf(endpoint, connection.getHeaderField("Location"))); - return new Response(read(connection.getInputStream()), - location == null ? null : endpoint.resolve(location)); + return read(connection.getInputStream(), MAX_BODY, false); } catch (IOException e) { throw new WebRtcException("Sending the offer to " + endpoint + " failed: " + e.getMessage(), e); @@ -216,19 +242,82 @@ private static void delete(URI resource, String bearerToken, Duration timeout) t HttpURLConnection connection = open(resource, "DELETE", bearerToken, timeout); try { - connection.getResponseCode(); + int status = connection.getResponseCode(); + + if (status / 100 != 2) { + throw new WebRtcException("The endpoint answered the deletion of " + resource + " with status " + + status + ": " + read(connection.getErrorStream(), MAX_ERROR_BODY, true)); + } } finally { connection.disconnect(); } } + /** + * Deletes a resource that nothing else will, and reports rather than + * throws if that fails. + */ + private static void discard(WebRtc rtc, URI resource, String bearerToken, Duration timeout) { + if (resource == null) { + return; + } + + try { + delete(resource, bearerToken, timeout); + } + catch (Exception e) { + rtc.reportError(new WebRtcException("Deleting the session " + resource + " failed", e)); + } + } + + private static void leftOnServer(WebRtc rtc, URI resource) { + if (resource != null) { + rtc.reportError(new WebRtcException("The session " + resource + + " was left on the server, since the runtime is closed")); + } + } + + /** + * Resolves the resource the server named for a session. It may be on + * another host, as WHIP and WHEP allow, but the bearer token goes along to + * delete it, so an https endpoint cannot name a plain http resource. + */ + static URI resourceOf(URI endpoint, String location) { + if (location == null) { + return null; + } + + URI resource; + + try { + resource = endpoint.resolve(location.trim()); + } + catch (IllegalArgumentException e) { + throw new WebRtcException("The endpoint " + endpoint + " named an invalid resource: " + location, e); + } + + String scheme = resource.getScheme(); + + if (!"http".equalsIgnoreCase(scheme) && !"https".equalsIgnoreCase(scheme)) { + throw new WebRtcException("The endpoint " + endpoint + " named a resource that is not http: " + resource); + } + if ("https".equalsIgnoreCase(endpoint.getScheme()) && !"https".equalsIgnoreCase(scheme)) { + throw new WebRtcException("The endpoint " + endpoint + " named a resource without TLS: " + resource); + } + + return resource; + } + private static HttpURLConnection open(URI uri, String method, String bearerToken, Duration timeout) throws IOException { HttpURLConnection connection = (HttpURLConnection) uri.toURL().openConnection(); connection.setRequestMethod(method); connection.setConnectTimeout((int) Math.min(Integer.MAX_VALUE, timeout.toMillis())); connection.setReadTimeout((int) Math.min(Integer.MAX_VALUE, timeout.toMillis())); + // A redirect would take the bearer token to wherever it points, so it + // fails the request instead. + connection.setInstanceFollowRedirects(false); if (bearerToken != null) { connection.setRequestProperty("Authorization", "Bearer " + bearerToken); @@ -237,7 +326,11 @@ private static HttpURLConnection open(URI uri, String method, String bearerToken return connection; } - private static String read(InputStream in) throws IOException { + /** + * Reads a response body of at most the given size. A larger one fails, or + * is cut off when only a message is made of it. + */ + private static String read(InputStream in, int limit, boolean truncate) throws IOException { if (in == null) { return ""; } @@ -248,6 +341,15 @@ private static String read(InputStream in) throws IOException { int count; while ((count = stream.read(buffer)) != -1) { + if (out.size() + count > limit) { + if (!truncate) { + throw new IOException("The response is larger than " + limit + " bytes"); + } + + out.write(buffer, 0, limit - out.size()); + break; + } + out.write(buffer, 0, count); } @@ -266,19 +368,4 @@ static URI checkEndpoint(URI endpoint) { return endpoint; } - - - - private static final class Response { - - final String body; - - final URI location; - - - Response(String body, URI location) { - this.body = body; - this.location = location; - } - } } diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java index 246d5fd8..1c8c8fa5 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java @@ -17,12 +17,16 @@ package dev.onvoid.webrtc.client; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.net.URI; import java.nio.ByteBuffer; +import java.time.Duration; +import java.util.Arrays; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; @@ -130,6 +134,111 @@ void refusedOfferLeavesNothing() throws Exception { } } + @Test + void lateAnswerDeleted() throws Exception { + // The answer arrives after the session timed out. The server created + // a resource for it all the same, which must not be left there. + try (TestHttpEndpoint endpoint = new TestHttpEndpoint(offer -> "", 201)) { + endpoint.stallMillis = 700; + + ExecutionException failed = assertThrows(ExecutionException.class, + () -> Whep.from(endpoint.uri()).timeout(Duration.ofSeconds(1)).play(rtc) + .get(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + assertTrue(failed.getCause().getMessage().contains("within"), failed.getCause().getMessage()); + awaitRequest(endpoint, "DELETE /live/session/1"); + } + } + + @Test + void oversizedAnswerRefused() throws Exception { + char[] huge = new char[2 << 20]; + Arrays.fill(huge, 'a'); + + try (TestHttpEndpoint endpoint = new TestHttpEndpoint(offer -> new String(huge), 201)) { + ExecutionException failed = assertThrows(ExecutionException.class, + () -> Whep.from(endpoint.uri()).play(rtc).get(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + assertTrue(failed.getCause().getMessage().contains("larger than"), failed.getCause().getMessage()); + // The resource was named before the answer was refused. + awaitRequest(endpoint, "DELETE /live/session/1"); + } + } + + @Test + void failedDeleteReported() throws Exception { + try (TestHttpEndpoint endpoint = receivingEndpoint()) { + endpoint.deleteStatus = 500; + + Whip.Session session = await(Whip.to(endpoint.uri()).track(video).publish(rtc)); + await(session.closeAsync()); + + assertDeleteReported("500"); + } + } + + @Test + void redirectNotFollowed() throws Exception { + // Following it would take the bearer token to wherever it points. + try (TestHttpEndpoint endpoint = receivingEndpoint()) { + endpoint.deleteStatus = 307; + + Whip.Session session = await(Whip.to(endpoint.uri()).bearerToken("secret").track(video).publish(rtc)); + await(session.closeAsync()); + + assertDeleteReported("307"); + assertFalse(endpoint.requests.stream().anyMatch(request -> request.path.equals(endpoint.redirect)), + () -> "Requests: " + endpoint.requests); + } + } + + @Test + void resourceKeepsTls() { + URI endpoint = URI.create("https://media.example/whep"); + + assertNull(HttpSession.resourceOf(endpoint, null)); + assertEquals(URI.create("https://media.example/session/1"), HttpSession.resourceOf(endpoint, "/session/1")); + // Another host is allowed, as WHIP and WHEP allow it. + assertEquals(URI.create("https://edge.example/session/1"), + HttpSession.resourceOf(endpoint, "https://edge.example/session/1")); + + assertThrows(WebRtcException.class, + () -> HttpSession.resourceOf(endpoint, "http://media.example/session/1")); + assertThrows(WebRtcException.class, + () -> HttpSession.resourceOf(URI.create("http://media.example/whep"), "file:/etc/passwd")); + } + + /** + * An endpoint whose side of the session receives what is published. + */ + private TestHttpEndpoint receivingEndpoint() throws Exception { + return new TestHttpEndpoint(offer -> answer(rtc.createPeer(PeerOptions.builder() + .negotiation(Negotiation.MANUAL) + .build()), offer), 201); + } + + /** + * Checks that the one error reported is the failed deletion, and takes it + * out of those that fail the test. + */ + private void assertDeleteReported(String status) { + assertEquals(1, errors.size(), () -> "Errors: " + errors); + assertTrue(errors.get(0).getCause().getMessage().contains(status), () -> "Errors: " + errors); + errors.clear(); + } + + private static void awaitRequest(TestHttpEndpoint endpoint, String request) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(TIMEOUT_SECONDS); + + while (endpoint.requests.stream().noneMatch(r -> r.toString().equals(request)) + && System.nanoTime() < deadline) { + Thread.sleep(50); + } + + assertTrue(endpoint.requests.stream().anyMatch(r -> r.toString().equals(request)), + () -> "Requests: " + endpoint.requests); + } + private static String answer(Peer peer, String offer) { try { return peer.negotiator() diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java index dc316ae0..439b3f28 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java @@ -33,7 +33,7 @@ /** * Just enough HTTP for WHIP and WHEP tests. A POST is answered with what the * handler returns for the offer, and creates a resource; a DELETE of that - * resource returns 200. + * resource returns 200, unless the test sets another status. */ final class TestHttpEndpoint implements AutoCloseable { @@ -61,6 +61,22 @@ public String toString() { final List requests = new CopyOnWriteArrayList<>(); + /** The Location a POST is answered with. */ + volatile String location = "/live/session/1"; + + /** The status a DELETE is answered with. */ + volatile int deleteStatus = 200; + + /** Where a DELETE answered with a redirect points. */ + volatile String redirect = "/live/moved"; + + /** + * How long the answer to a POST pauses, twice: before its status line and + * after it. Each pause stays within a client's read timeout while both + * together may not. + */ + volatile long stallMillis; + private final ServerSocket server; private final Function answerer; @@ -133,28 +149,46 @@ else if (name.equals("authorization")) { if (requestLine[0].equals("POST")) { if (status != 201) { - respond(out, status + " Refused", "", "not allowed"); + respond(out, status + " Refused", "", "not allowed", 0); return; } String answer = answerer.apply(new String(body, StandardCharsets.UTF_8)); - respond(out, "201 Created", "Location: /live/session/1\r\n", answer); + respond(out, "201 Created", "Location: " + location + "\r\n", answer, stallMillis); + } + else if (deleteStatus / 100 == 3) { + respond(out, deleteStatus + " Moved", "Location: " + redirect + "\r\n", "", 0); } else { - respond(out, "200 OK", "", ""); + respond(out, deleteStatus + " Status", "", "", 0); } } - private static void respond(OutputStream out, String status, String headers, String body) throws IOException { + private static void respond(OutputStream out, String status, String headers, String body, long stallMillis) + throws IOException { byte[] bytes = body.getBytes(StandardCharsets.UTF_8); - String head = "HTTP/1.1 " + status + "\r\nContent-Type: application/sdp\r\n" + headers + String head = "Content-Type: application/sdp\r\n" + headers + "Content-Length: " + bytes.length + "\r\nConnection: close\r\n\r\n"; + stall(stallMillis); + out.write(("HTTP/1.1 " + status + "\r\n").getBytes(StandardCharsets.US_ASCII)); + out.flush(); + stall(stallMillis); out.write(head.getBytes(StandardCharsets.US_ASCII)); out.write(bytes); out.flush(); } + private static void stall(long millis) throws IOException { + try { + Thread.sleep(millis); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + private static String readLine(InputStream in) throws IOException { ByteArrayOutputStream line = new ByteArrayOutputStream(); int c;