diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt index 4f84a6876b..6b971927fb 100644 --- a/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt +++ b/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt @@ -85,11 +85,13 @@ import org.jitsi.videobridge.relay.RelayedEndpoint import org.jitsi.videobridge.rest.root.debug.EndpointDebugFeatures import org.jitsi.videobridge.stats.PacketTransitStats import org.jitsi.videobridge.transport.dtls.DtlsTransport +import org.jitsi.videobridge.transport.ice.IceRestartResult import org.jitsi.videobridge.transport.ice.IceTransport import org.jitsi.videobridge.util.ByteBufferPool import org.jitsi.videobridge.util.TaskPools import org.jitsi.videobridge.util.looksLikeDtls import org.jitsi.videobridge.websocket.colibriWebSocketServiceSupplier +import org.jitsi.videobridge.websocket.generateColibriWebSocketPassword import org.jitsi.xmpp.extensions.colibri.WebSocketPacketExtension import org.jitsi.xmpp.extensions.jingle.DtlsFingerprintPacketExtension import org.jitsi.xmpp.extensions.jingle.IceUdpTransportPacketExtension @@ -170,6 +172,13 @@ class Endpoint @JvmOverloads constructor( } } + /** + * The password which authenticates the colibri WebSocket of this endpoint. It is independent of ICE and + * stays the same for the lifetime of the endpoint, because the client re-dials the URL which contains it + * each time the WebSocket reconnects. See [generateColibriWebSocketPassword]. + */ + private val webSocketPassword = generateColibriWebSocketPassword() + /* TODO: do we ever want to support useUniquePort for an Endpoint? */ private val iceTransport = IceTransport(id, iceControlling, false, supportsPrivateAddresses, logger) private val dtlsTransport = DtlsTransport(logger, id).also { it.cryptex = CryptexConfig.endpoint } @@ -731,11 +740,8 @@ class Endpoint @JvmOverloads constructor( * @return {@code true} iff the password matches. */ fun acceptWebSocket(password: String): Boolean { - if (iceTransport.icePassword != password) { - logger.warn( - "Incoming web socket request with an invalid password. " + - "Expected: ${iceTransport.icePassword} received $password" - ) + if (webSocketPassword != password) { + logger.warn("Incoming web socket request with an invalid password.") return false } return true @@ -786,6 +792,19 @@ class Endpoint @JvmOverloads constructor( iceTransport.startConnectivityEstablishment(transportInfo) } + /** + * Handles an explicit ICE restart request from this endpoint (colibri2 ``). + * + * Creates a new ice4j Agent with freshly rotated local credentials alongside the established one, which + * keeps carrying media until the new one connects (make-before-break). + * + * @return what the caller must signal back to the endpoint: our new transport (returned by + * [describeTransport], which then describes the pending Agent) for [IceRestartResult.STARTED], the + * unchanged established transport for [IceRestartResult.KEEP_EXISTING], or no transport at all for + * [IceRestartResult.UNAVAILABLE]. + */ + fun requestIceRestart(): IceRestartResult = iceTransport.requestIceRestart() + fun describeTransport(): IceUdpTransportPacketExtension { val iceUdpTransportPacketExtension = IceUdpTransportPacketExtension() iceTransport.describe(iceUdpTransportPacketExtension) @@ -794,7 +813,7 @@ class Endpoint @JvmOverloads constructor( colibriWebsocketService.getColibriWebSocketUrls( conference.id, id, - iceTransport.icePassword + webSocketPassword ).forEach { wsUrl -> val wsPacketExtension = WebSocketPacketExtension(wsUrl) iceUdpTransportPacketExtension.addChildExtension(wsPacketExtension) diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/colibri2/Colibri2ConferenceHandler.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/colibri2/Colibri2ConferenceHandler.kt index 4f257b52da..f5d2b320e6 100644 --- a/jvb/src/main/kotlin/org/jitsi/videobridge/colibri2/Colibri2ConferenceHandler.kt +++ b/jvb/src/main/kotlin/org/jitsi/videobridge/colibri2/Colibri2ConferenceHandler.kt @@ -31,6 +31,7 @@ import org.jitsi.videobridge.relay.AudioSourceDesc import org.jitsi.videobridge.relay.Relay import org.jitsi.videobridge.relay.RelayConfig import org.jitsi.videobridge.sctp.SctpConfig +import org.jitsi.videobridge.transport.ice.IceRestartResult import org.jitsi.videobridge.util.PayloadTypeUtil.Companion.create import org.jitsi.videobridge.websocket.config.WebsocketServiceConfig import org.jitsi.videobridge.xmpp.MediaSourceFactory @@ -257,19 +258,6 @@ class Colibri2ConferenceHandler( } c2endpoint.transport?.iceUdpTransport?.let { endpoint.setTransportInfo(it) } - if (c2endpoint.create) { - val transBuilder = Transport.getBuilder() - transBuilder.setIceUdpExtension(endpoint.describeTransport()) - if (c2endpoint.transport?.sctp != null) { - transBuilder.setSctp( - Sctp.Builder() - .setPort(DcSctpTransport.DEFAULT_SCTP_PORT) - .setRole(Sctp.Role.SERVER) - .build() - ) - } - respBuilder.setTransport(transBuilder.build()) - } c2endpoint.sources?.let { sources -> if (endpoint.visitor && sources.mediaSources.isNotEmpty()) { @@ -309,6 +297,50 @@ class Colibri2ConferenceHandler( endpoint.updateForceMute(it.audio, it.video) } + // An explicit ICE restart request. Handled after setTransportInfo, so any credentials included in this + // same request still belong to (and are applied to) the pre-restart Agent. The restart rotates our own + // ICE credentials, so the endpoint has to be told the new ones — its connectivity checks are addressed + // to them — even though this is not a create. + // + // Handled last, once nothing else in this request can throw: a restart is armed as a side effect (it + // rotates the credentials we advertise and starts the timeout), so failing the request after arming it + // would leave the bridge waiting out that timeout for an answer the endpoint was never asked for. + val iceRestartResult = if (c2endpoint.transport?.iceRestart == true) { + endpoint.requestIceRestart().also { + if (it != IceRestartResult.STARTED) { + logger.warn("Did not restart ICE for endpoint ${c2endpoint.id}: $it.") + } + } + } else { + null + } + + // How the refusal of a restart is signaled: with the *absence* of a in the + // conference-modified for this endpoint. There is no explicit "refused" flag, and the request is not + // failed with an error, because an error would fail the whole conference-modify and take every other + // endpoint's updates with it. Jicofo pairs a request it sent with the answer it gets back: a + // means the restart happened and is relayed to the endpoint, no means it did + // not and jicofo falls back to a re-invite. + // + // So a transport is signaled back for a restart that started (the new Agent's rotated credentials) and + // for one that kept the existing Agent (its unchanged credentials, so the endpoint keeps the connection + // it has, with no re-invite). Only IceRestartResult.UNAVAILABLE signals nothing. + if (c2endpoint.create || iceRestartResult == IceRestartResult.STARTED || + iceRestartResult == IceRestartResult.KEEP_EXISTING + ) { + val transBuilder = Transport.getBuilder() + transBuilder.setIceUdpExtension(endpoint.describeTransport()) + if (c2endpoint.transport?.sctp != null) { + transBuilder.setSctp( + Sctp.Builder() + .setPort(DcSctpTransport.DEFAULT_SCTP_PORT) + .setRole(Sctp.Role.SERVER) + .build() + ) + } + respBuilder.setTransport(transBuilder.build()) + } + return respBuilder.build() } diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/ice/IceConfig.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/ice/IceConfig.kt index fc68992ff5..2119e8b6ed 100644 --- a/jvb/src/main/kotlin/org/jitsi/videobridge/ice/IceConfig.kt +++ b/jvb/src/main/kotlin/org/jitsi/videobridge/ice/IceConfig.kt @@ -22,6 +22,7 @@ import org.jitsi.config.JitsiConfig import org.jitsi.metaconfig.config import org.jitsi.metaconfig.from import org.jitsi.metaconfig.optionalconfig +import java.time.Duration class IceConfig private constructor() { /** @@ -70,6 +71,34 @@ class IceConfig private constructor() { "videobridge.ice.advertise-private-candidates".from(JitsiConfig.newConfig) ) + /** + * Whether ICE restarts are enabled: when an endpoint explicitly requests one (colibri2 + * ``), create a second [org.ice4j.ice.Agent] with rotated local credentials + * and run it alongside the established one, instead of rejecting the request. + */ + val restartEnabled: Boolean by config( + "videobridge.ice.restart.enabled".from(JitsiConfig.newConfig) + ) + + /** + * How long the old [org.ice4j.ice.Agent] is kept alive after an ICE restart has cut over to the new one. + * Both Agents keep their sockets during this window, and ice4j routes each packet by the address it came + * from, so the endpoint's old-generation checks — which come from its old address — are answered by the old + * Agent rather than dropped. + */ + val restartTransitionWindow: Duration by config( + "videobridge.ice.restart.transition-window".from(JitsiConfig.newConfig) + ) + + /** + * How long to wait for the new [org.ice4j.ice.Agent] of an ICE restart to connect before giving up on the + * restart and keeping the existing Agent. A value that is not positive disables ICE restarts: the new Agent + * would be freed before the endpoint could answer it. + */ + val restartTimeout: Duration by config( + "videobridge.ice.restart.timeout".from(JitsiConfig.newConfig) + ) + companion object { @JvmField val config = IceConfig() diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/relay/Relay.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/relay/Relay.kt index 88c1348069..cfbc4902cf 100644 --- a/jvb/src/main/kotlin/org/jitsi/videobridge/relay/Relay.kt +++ b/jvb/src/main/kotlin/org/jitsi/videobridge/relay/Relay.kt @@ -97,6 +97,7 @@ import org.jitsi.videobridge.util.ByteBufferPool import org.jitsi.videobridge.util.TaskPools import org.jitsi.videobridge.util.looksLikeDtls import org.jitsi.videobridge.websocket.colibriWebSocketServiceSupplier +import org.jitsi.videobridge.websocket.generateColibriWebSocketPassword import org.jitsi.xmpp.extensions.colibri.WebSocketPacketExtension import org.jitsi.xmpp.extensions.colibri2.Sctp import org.jitsi.xmpp.extensions.jingle.DtlsFingerprintPacketExtension @@ -201,6 +202,13 @@ class Relay @JvmOverloads constructor( } } + /** + * The password which authenticates the colibri WebSocket of this relay. It is independent of ICE and stays + * the same for the lifetime of the relay, because the peer re-dials the URL which contains it each time the + * WebSocket reconnects. See [generateColibriWebSocketPassword]. + */ + private val webSocketPassword = generateColibriWebSocketPassword() + private val iceTransport = IceTransport( id = id, controlling = iceControlling, @@ -540,7 +548,7 @@ class Relay @JvmOverloads constructor( val urls = colibriWebsocketService.getColibriRelayWebSocketUrls( conference.id, id, - iceTransport.icePassword + webSocketPassword ) if (urls.isEmpty()) { logger.warn("No colibri relay URLs configured") @@ -872,11 +880,8 @@ class Relay @JvmOverloads constructor( * @return {@code true} iff the password matches. */ fun acceptWebSocket(password: String): Boolean { - if (iceTransport.icePassword != password) { - logger.warn( - "Incoming web socket request with an invalid password. " + - "Expected: ${iceTransport.icePassword} received $password" - ) + if (webSocketPassword != password) { + logger.warn("Incoming web socket request with an invalid password.") return false } return true diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/transport/ice/IceTransport.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/transport/ice/IceTransport.kt index 397b480f10..8407c603b3 100755 --- a/jvb/src/main/kotlin/org/jitsi/videobridge/transport/ice/IceTransport.kt +++ b/jvb/src/main/kotlin/org/jitsi/videobridge/transport/ice/IceTransport.kt @@ -23,6 +23,7 @@ import org.ice4j.Transport import org.ice4j.TransportAddress import org.ice4j.ice.Agent import org.ice4j.ice.CandidateType +import org.ice4j.ice.Component import org.ice4j.ice.HostCandidate import org.ice4j.ice.IceMediaStream import org.ice4j.ice.IceProcessingState @@ -52,10 +53,34 @@ import java.io.IOException import java.net.DatagramPacket import java.net.Inet6Address import java.time.Clock +import java.time.Duration import java.time.Instant +import java.util.concurrent.ScheduledFuture +import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.LongAdder +/** + * The outcome of an ICE restart request, which decides what the caller signals back to the peer. + */ +enum class IceRestartResult { + /** A new Agent was created. Signal its transport, so the peer moves its checks to it. */ + STARTED, + + /** + * No new Agent was created, but the established one is still usable. Signal its transport unchanged: the + * peer keeps the connection it has, instead of escalating to a full re-invite over a transport that is + * merely still connecting. + */ + KEEP_EXISTING, + + /** + * This bridge can not restart ICE. Signal no transport at all, which is how the peer learns to fall back to + * a full re-invite. + */ + UNAVAILABLE +} + class IceTransport @JvmOverloads constructor( id: String, /** @@ -73,10 +98,25 @@ class IceTransport @JvmOverloads constructor( */ private val advertisePrivateAddresses: Boolean, parentLogger: Logger, - private val clock: Clock = Clock.systemUTC() + private val clock: Clock = Clock.systemUTC(), + /** + * Creates the ice4j [Agent]s of this transport, with their candidate harvesters configured. Only replaced + * in tests, which have no use for real Agents (they bind ports and start threads). + */ + agentFactory: ((Logger) -> Agent)? = null ) { private val logger = createChildLogger(parentLogger) + private val agentFactory: (Logger) -> Agent = agentFactory ?: { agentLogger -> + Agent(IceConfig.config.ufragPrefix, agentLogger).apply { + if (useUniquePort) { + setUseDynamicPorts(true) + } else { + appendHarvesters(this) + } + } + } + /** * The handler which will be invoked when data is received. * This field should be set by some other entity which wishes to handle the incoming data @@ -123,40 +163,157 @@ class IceTransport @JvmOverloads constructor( */ private val running = AtomicBoolean(true) - private val iceStateChangeListener = PropertyChangeListener { ev -> iceStateChanged(ev) } - private val iceStreamPairChangedListener = PropertyChangeListener { ev -> iceStreamPairChanged(ev) } + private val packetStats = PacketStats() - private val iceAgent = Agent(IceConfig.config.ufragPrefix, logger).apply { - if (useUniquePort) { - setUseDynamicPorts(true) - } else { - appendHarvesters(this) - } - isControlling = controlling - performConsentFreshness = true - nominationStrategy = IceConfig.config.nominationStrategy - addStateChangeListener(iceStateChangeListener) - }.also { - logger.addContext("local_ufrag", it.localUfrag) + /** + * Whether the ice4j [Agent]s created by this transport take the 'controlling' role. + */ + private val controlling = controlling + + /** + * Guards transitions between [currentBundle] and [pendingBundle]. + */ + private val restartLock = Any() + + /** + * The [AgentBundle] that media is currently sent on and whose credentials we advertise. Replaced (via + * [cutOver]) when an ICE restart's new Agent connects. + */ + @Volatile + private var currentBundle: AgentBundle + + /** + * The [AgentBundle] of an ICE restart that is in progress but has not connected yet, if any. It runs + * alongside [currentBundle], which keeps sending until the cutover (make-before-break). + */ + @Volatile + private var pendingBundle: AgentBundle? = null + + /** + * The [AgentBundle] we cut over from, for as long as its transition window lasts. Held here, and not only + * by the task that frees it, so that [stop] can free it right away instead of leaving an Agent alive after + * the transport is gone. + * + * Guarded by [restartLock]. + */ + private var retiringBundle: AgentBundle? = null + + /** + * The task that abandons [pendingBundle] if it does not connect in time, if one is scheduled. + * + * Guarded by [restartLock]. + */ + private var restartTimeoutTask: ScheduledFuture<*>? = null + + /** + * The task that frees [retiringBundle] when its transition window elapses, if one is scheduled. + * + * Guarded by [restartLock]. + */ + private var transitionWindowTask: ScheduledFuture<*>? = null + + /** + * The ICE generation of the most recent restart we started. The bridge owns this counter: each accepted + * restart request gets the next value, it is advertised on the transport we return (as `ice-generation`), + * and the peer echoes it on the transport-info carrying its own new credentials. That lets us match the + * peer's response to the right pending bundle and discard responses from a superseded round. + * + * Restart generations start at 1. Generation 0 means "the initial allocation, never restarted" and + * [IceUdpTransportPacketExtension.GENERATION_UNSPECIFIED] means the attribute is absent altogether: the initial + * bundle carries the latter, so nothing is stamped on the wire until a restart actually happens. Clients reject + * a restart transport whose generation is not >= 1, so this must not start any lower. + * + * Guarded by [restartLock]. + */ + private var restartGeneration = 0 + + init { + currentBundle = createAgentBundle(IceUdpTransportPacketExtension.GENERATION_UNSPECIFIED) + ?: throw IllegalStateException("Failed to create the initial ICE Agent") + logger.addContext("local_ufrag", currentBundle.agent.localUfrag) } - private val iceStream = iceAgent.createMediaStream("stream").apply { - addPairChangeListener(iceStreamPairChangedListener) + /** + * Creates an [Agent] with its stream and component, and wraps them in an [AgentBundle]. + * + * @return the new bundle, or null if ice4j failed to create it (creating a component harvests candidates + * and can fail to bind). Anything that was already created is freed, so a failure leaks neither an Agent + * nor its ufrag in the single port harvester. + */ + private fun createAgentBundle(generation: Int): AgentBundle? { + var agent: Agent? = null + return try { + val newAgent = agentFactory(logger).apply { + isControlling = this@IceTransport.controlling + performConsentFreshness = true + nominationStrategy = IceConfig.config.nominationStrategy + } + agent = newAgent + val stream = newAgent.createMediaStream("stream") + val component = newAgent.createComponent(stream, IceConfig.config.keepAliveStrategy, false) + AgentBundle(generation, newAgent, stream, component) + } catch (t: Throwable) { + logger.error("Failed to create an ICE Agent (generation=$generation).", t) + agent?.free() + null + } } - private val iceComponent = iceAgent.createComponent(iceStream, IceConfig.config.keepAliveStrategy, false).apply { - setBufferCallback(object : BufferHandler { - override fun handleBuffer(buffer: Buffer) { - incomingDataHandler?.dataReceived(buffer) ?: run { - packetStats.numIncomingPacketsDroppedNoHandler.increment() - ByteBufferPool.returnBuffer(buffer.buffer) + /** + * An ice4j [Agent] together with the single stream and component we create on it and the listeners we + * attach to them. An [IceTransport] has exactly one of these normally, and briefly two while an ICE + * restart is in flight: the established one (which keeps sending) and the new one (which is running + * connectivity checks with freshly rotated local credentials). + * + * Created by [createAgentBundle], which is also where a failure to create the ice4j objects is handled. + * + * @param generation the `ice-generation` of the restart round that created this bundle, or + * [IceUdpTransportPacketExtension.GENERATION_UNSPECIFIED] for the initial bundle. + */ + private inner class AgentBundle( + val generation: Int, + val agent: Agent, + val stream: IceMediaStream, + val component: Component + ) { + private val stateChangeListener = PropertyChangeListener { ev -> iceStateChanged(this@AgentBundle, ev) } + private val pairChangeListener = PropertyChangeListener { ev -> iceStreamPairChanged(this@AgentBundle, ev) } + + /** When this bundle was created, used to report how long a restart took. */ + val createdAt: Instant = clock.instant() + + /** + * Whether connectivity establishment has been started on [agent]. For a restart's bundle this only + * happens once the peer's new remote credentials arrive, and it must happen at most once. + */ + val checksStarted = AtomicBoolean(false) + + init { + agent.addStateChangeListener(stateChangeListener) + stream.addPairChangeListener(pairChangeListener) + component.setBufferCallback(object : BufferHandler { + override fun handleBuffer(buffer: Buffer) { + incomingDataHandler?.dataReceived(buffer) ?: run { + packetStats.numIncomingPacketsDroppedNoHandler.increment() + ByteBufferPool.returnBuffer(buffer.buffer) + } } + }) + } + + private val freed = AtomicBoolean(false) + + fun free() { + if (!freed.compareAndSet(false, true)) { + return } - }) + agent.removeStateChangeListener(stateChangeListener) + stream.removePairStateChangeListener(pairChangeListener) + agent.free() + } + + override fun toString(): String = "AgentBundle[generation=$generation, ufrag=${agent.localUfrag}]" } - private val packetStats = PacketStats() - val icePassword: String - get() = iceAgent.localPassword /** * Tell this [IceTransport] to start ICE connectivity establishment. @@ -166,19 +323,42 @@ class IceTransport @JvmOverloads constructor( logger.warn("Not starting connectivity establishment, transport is not running") return } - if (iceAgent.state.isEstablished) { + + // An update tagged with an `ice-generation` is the peer answering an ICE restart with the new + // credentials that our new Agent needs in order to address its own connectivity checks. Route it to the + // pending bundle instead of the established one. + // + // Only a tagged update: the peer stamps the generation on the restart answer and on nothing else, so an + // untagged update arriving while a restart is in flight is an ordinary transport update (a trickled + // candidate, or an old one still on the wire) and belongs to the established bundle. Applying such an + // update to the pending bundle would give it the peer's *old* password, whose checks the peer rejects. + // + // Note that the peer's own trickled candidates never reach the pending bundle this way, and the + // candidates of a tagged update never reach the established one. Neither matters for the bridge: it + // signals no candidates of its own that the peer must answer, and it discovers the peer's address + // peer-reflexively from the peer's incoming checks. + val pending = pendingBundle + if (pending != null && + transportPacketExtension.iceGeneration != IceUdpTransportPacketExtension.GENERATION_UNSPECIFIED + ) { + applyRemoteCredentialsToPendingRestart(pending, transportPacketExtension) + return + } + + val bundle = currentBundle + if (bundle.agent.state.isEstablished) { logger.cdebug { "Connection already established" } return } logger.cdebug { "Starting ICE connectivity establishment" } // Set the remote ufrag/password - iceStream.remoteUfrag = transportPacketExtension.ufrag - iceStream.remotePassword = transportPacketExtension.password + bundle.stream.remoteUfrag = transportPacketExtension.ufrag + bundle.stream.remotePassword = transportPacketExtension.password // If ICE is running already, we try to update the checklists with the // candidates. Note that this is a best effort. - val iceAgentStateIsRunning = IceProcessingState.RUNNING == iceAgent.state + val iceAgentStateIsRunning = IceProcessingState.RUNNING == bundle.agent.state val remoteCandidates = transportPacketExtension.getChildExtensionsOfType(CandidatePacketExtension::class.java) if (iceAgentStateIsRunning && remoteCandidates.isEmpty()) { @@ -189,7 +369,7 @@ class IceTransport @JvmOverloads constructor( return } - val remoteCandidateCount = addRemoteCandidates(remoteCandidates, iceAgentStateIsRunning) + val remoteCandidateCount = addRemoteCandidates(bundle, remoteCandidates, iceAgentStateIsRunning) if (iceAgentStateIsRunning) { when (remoteCandidateCount) { 0 -> { @@ -197,7 +377,7 @@ class IceTransport @JvmOverloads constructor( // candidates were ignored: // iceAgentStateIsRunning && candidates.isEmpty(). } - else -> iceComponent.updateRemoteCandidates() + else -> bundle.component.updateRemoteCandidates() } } else if (remoteCandidateCount != 0) { // Once again, because the ICE Agent does not support adding @@ -206,23 +386,334 @@ class IceTransport @JvmOverloads constructor( // the whole set of transport candidates from the remote peer to the // local peer, do not really start the connectivity establishment // until we have at least one remote candidate per ICE Component. - if (iceComponent.remoteCandidateCount > 0) { + if (bundle.component.remoteCandidateCount > 0) { logger.debug("Starting the agent with remote candidates.") - iceAgent.startConnectivityEstablishment() + bundle.agent.startConnectivityEstablishment() + bundle.checksStarted.set(true) } - } else if (iceStream.remoteUfragAndPasswordKnown()) { + } else if (bundle.stream.remoteUfragAndPasswordKnown()) { // We don't have any remote candidates, but we already know the // remote ufrag and password, so we can start ICE. logger.debug("Starting the Agent without remote candidates.") - iceAgent.startConnectivityEstablishment() + bundle.agent.startConnectivityEstablishment() + bundle.checksStarted.set(true) } else { logger.cdebug { "Not starting ICE, no ufrag and pwd yet. ${transportPacketExtension.toXML()}" } } } - fun startReadingData() { + /** + * Starts an ICE restart, as explicitly requested by the peer (colibri2 ``). + * + * We create a second [Agent] with freshly rotated local credentials of our own (RFC 8445 section 9, so that + * checks belonging to different restart generations can be told apart by their credentials) and assign it + * the next [restartGeneration]. Connectivity checks are deliberately *not* started yet: we are the + * controlling agent, so our own checks need the peer's new remote credentials to build their USERNAME and + * MESSAGE-INTEGRITY, and those only arrive later (see [applyRemoteCredentialsToPendingRestart]). The new + * Agent does answer incoming checks in the meantime — ice4j starts its connectivity check server when the + * component is created and queues pre-RUNNING checks until we start. + * + * The established Agent keeps its selected pair and keeps sending throughout, so media is not interrupted + * (make-before-break); we only [cutOver] once the new Agent connects, and keep the old one alive for + * [IceConfig.restartTransitionWindow] after that so old-generation checks still in flight are answered. + * + * This assumes the peer's address changes, which is what a restart is for: ice4j's single port harvester + * demultiplexes on the remote address (see [cutOver]), so the new Agent is only reached by checks from an + * address the old Agent's socket is not already bound to. Checks from an unchanged address are delivered to + * the old Agent instead, which drops them because their local ufrag is not its own, and the new Agent — + * which has no signalled remote candidates and discovers the peer peer-reflexively — never connects. The + * restart is then abandoned and the established Agent is kept, which is the same outcome as any other + * failed restart. + * + * @return [IceRestartResult.STARTED] if a restart was started, in which case the caller must signal our new + * transport (rotated credentials plus the new `ice-generation`) back to the peer. + * [IceRestartResult.KEEP_EXISTING] if no restart was started but the established transport is still usable, + * in which case the caller must signal that transport unchanged. [IceRestartResult.UNAVAILABLE] if this + * bridge can not restart ICE at all, in which case the caller must signal no transport so that the peer + * falls back to a full re-invite. + */ + fun requestIceRestart(): IceRestartResult { + if (!running.get()) { + logger.warn("Can not restart ICE: the transport is not running.") + iceRestartsRejected.inc() + return IceRestartResult.UNAVAILABLE + } + if (!IceConfig.config.restartEnabled) { + logger.warn( + "Can not restart ICE: ICE restarts are disabled (videobridge.ice.restart.enabled=false)." + ) + iceRestartsRejected.inc() + return IceRestartResult.UNAVAILABLE + } + val timeout = IceConfig.config.restartTimeout + if (timeout.isZero || timeout.isNegative) { + // The restart would be abandoned before the peer could even answer, and the peer would be left + // holding credentials of an Agent we had already freed. Treat it the same as being disabled. + logger.warn( + "Can not restart ICE: videobridge.ice.restart.timeout is not positive ($timeout)." + ) + iceRestartsRejected.inc() + return IceRestartResult.UNAVAILABLE + } + if (!currentBundle.agent.state.isEstablished) { + // There is no established connectivity to preserve, so there is nothing for a make-before-break + // restart to do. The initial Agent is still gathering/checking and the peer should keep using it. + logger.warn( + "Not restarting ICE: the transport is not established yet " + + "(state=${currentBundle.agent.state}). Keeping the existing Agent." + ) + iceRestartsNotEstablished.inc() + return IceRestartResult.KEEP_EXISTING + } + + val newBundle: AgentBundle + val generation: Int + synchronized(restartLock) { + // A repeated request for a restart that has not started its checks yet (the peer has not answered + // with its own credentials) is answered with the bundle we already have. Clients do fire duplicate + // network-change events, and rolling a new generation here would free a bundle the peer may already + // be checking against, discard its progress and restart the timeout. + pendingBundle?.let { pending -> + if (!pending.checksStarted.get()) { + logger.info("An ICE restart is already pending and has not started checks: $pending.") + return IceRestartResult.STARTED + } + } + + // Create the new Agent before touching any state, so that a failure leaves the restart in flight + // (if there is one) alone rather than freeing a bundle the peer may already be checking against. + generation = restartGeneration + 1 + newBundle = createAgentBundle(generation) ?: run { + // A resource problem (no port to bind, most likely), so this bridge can not restart ICE at all + // right now. Escalate rather than keep the endpoint on a transport it just told us it can no + // longer reach: the established Agent still works, but its path is probably already dead. + logger.error("Can not restart ICE: failed to create the new Agent.") + iceRestartsAgentCreationFailed.inc() + return IceRestartResult.UNAVAILABLE + } + + // A newer restart supersedes one that has not connected yet. + pendingBundle?.let { superseded -> + logger.info("Superseding an in-flight ICE restart: $superseded") + iceRestartsSuperseded.inc() + TaskPools.IO_POOL.submit { superseded.free() } + } + restartTimeoutTask?.cancel(false) + restartTimeoutTask = null + + restartGeneration = generation + pendingBundle = newBundle + } + + logger.info( + "ICE restart requested (generation=$generation): created a pending Agent with local ufrag=" + + "${newBundle.agent.localUfrag} (current local ufrag=${currentBundle.agent.localUfrag}). " + + "Waiting for the peer's new remote credentials before starting connectivity checks." + ) + iceRestartsStarted.inc() + + val timeoutTask = TaskPools.SCHEDULED_POOL.schedule( + { abandonPendingRestart(newBundle, "it did not connect within $timeout") }, + timeout.toMillis(), + TimeUnit.MILLISECONDS + ) + synchronized(restartLock) { + // Unless the restart is already over, in which case there is nothing left to abandon. + if (pendingBundle === newBundle) { + restartTimeoutTask = timeoutTask + } else { + timeoutTask.cancel(false) + } + } + + return IceRestartResult.STARTED + } + + /** + * Handles a generation-tagged transport update that arrived while an ICE restart is pending. This is step + * two of a restart: the peer has applied the transport we returned from [requestIceRestart] and is now + * signalling its own new ICE credentials, tagged with the `ice-generation` we advertised. Apply them to the + * pending Agent and start its connectivity checks, which is the first moment we can — the checks we send as + * the controlling agent are authenticated with the peer's password. + */ + private fun applyRemoteCredentialsToPendingRestart( + pending: AgentBundle, + transportPacketExtension: IceUdpTransportPacketExtension + ) { + val generation = transportPacketExtension.iceGeneration + if (generation != pending.generation) { + // The peer answered a round we have since moved on from. This is the normal outcome of a superseded + // restart, not a rejected request, so it is not counted as one. + logger.info( + "Ignoring remote credentials for ICE generation $generation, the pending ICE restart is " + + "generation ${pending.generation}." + ) + return + } + + val ufrag = transportPacketExtension.ufrag + val password = transportPacketExtension.password + if (ufrag == null || password == null) { + logger.warn( + "Ignoring a transport update with no ufrag/pwd while ICE restart generation " + + "${pending.generation} is pending." + ) + return + } + + // Claim the round before applying anything. Everything that can make an update unusable has been + // checked by now (the credentials are present, they carry our generation, and the restart is still + // pending), so claiming here can not burn the round on an update we then reject. Applying first would + // instead let a repeated update overwrite the credentials of an Agent whose checks are already running: + // ice4j reads them live off the stream to sign its checks and to validate the peer's + // (ConnectivityCheckServer.getRemoteKey). + synchronized(restartLock) { + // Re-check under the lock: the restart may have been superseded or abandoned since we read it. + if (pendingBundle !== pending) { + logger.info( + "ICE restart generation ${pending.generation} is no longer pending, dropping the " + + "remote credentials that arrived for it." + ) + return + } + if (!pending.checksStarted.compareAndSet(false, true)) { + logger.info( + "Already started connectivity checks for ICE restart generation ${pending.generation}, " + + "ignoring a repeated transport update (remote ufrag=$ufrag)." + ) + return + } + pending.stream.remoteUfrag = ufrag + pending.stream.remotePassword = password + } + + // Add any signalled remote candidates. Normally there are none: clients do not signal candidates to the + // bridge and are discovered peer-reflexively from their incoming checks, which is also how a new address + // after a real network change is learned. + val remoteCandidates = transportPacketExtension.getChildExtensionsOfType(CandidatePacketExtension::class.java) + val remoteCandidateCount = addRemoteCandidates(pending, remoteCandidates, iceAgentIsRunning = false) + + logger.info( + "Applied the peer's new remote credentials to the pending ICE restart " + + "(generation=${pending.generation}, remote ufrag=$ufrag, remoteCandidates=" + + "$remoteCandidateCount, local ufrag=${pending.agent.localUfrag}), starting connectivity checks." + ) + pending.agent.startConnectivityEstablishment() + } + + /** + * Switches [currentBundle] over to [newBundle], which has just connected, and schedules the old bundle to be + * freed after [IceConfig.restartTransitionWindow]. + * + * Both bundles keep their sockets during that window, and both keep accepting whatever arrives on them: + * connectivity checks, and media. What routes a packet to one or the other is the peer's address, not the + * local ufrag it carries — ice4j's single port harvester looks the remote address up in its socket map + * first and only parses the ufrag for an address it has never seen (`AbstractUdpListener`). So the window + * is useful because the peer's old-generation checks come *from the old address*, which is still mapped to + * the old Agent's socket, and are answered there rather than dropped. See [requestIceRestart] for what this + * demultiplexing means for the new Agent. + */ + private fun cutOver(newBundle: AgentBundle) { + val transitionWindow = IceConfig.config.restartTransitionWindow + val keepOldBundle = !transitionWindow.isZero && !transitionWindow.isNegative + val oldBundle = synchronized(restartLock) { + if (pendingBundle !== newBundle) { + // Superseded or abandoned while it was connecting. + return + } + val old = currentBundle + currentBundle = newBundle + pendingBundle = null + restartTimeoutTask?.cancel(false) + restartTimeoutTask = null + + // A bundle still retiring from an earlier cutover has been superseded twice over by now. Free it + // rather than let two old Agents linger. + retiringBundle?.let { previous -> + transitionWindowTask?.cancel(false) + transitionWindowTask = null + TaskPools.IO_POOL.submit { previous.free() } + } + retiringBundle = if (keepOldBundle) old else null + old + } + + val elapsedMs = Duration.between(newBundle.createdAt, clock.instant()).toMillis() + logger.info( + "ICE restart (generation=${newBundle.generation}) connected after ${elapsedMs}ms, cutting over " + + "from local ufrag ${oldBundle.agent.localUfrag} to ${newBundle.agent.localUfrag}. Freeing the " + + "old Agent in $transitionWindow." + ) + iceRestartsSucceeded.inc() + + if (useUniquePort) { + // ice4j's push API only works with the single port harvester, so with unique ports we have to read + // from the new Agent's socket ourselves. The old bundle's reader exits when its socket is closed. + TaskPools.IO_POOL.submit { startReadingData(newBundle) } + } + + if (!keepOldBundle) { + TaskPools.IO_POOL.submit { oldBundle.free() } + return + } + + val freeTask = TaskPools.SCHEDULED_POOL.schedule( + { + logger.info( + "ICE restart (generation=${newBundle.generation}) transition window elapsed, freeing " + + "the old Agent with local ufrag ${oldBundle.agent.localUfrag}." + ) + synchronized(restartLock) { + if (retiringBundle === oldBundle) { + retiringBundle = null + transitionWindowTask = null + } + } + // Not inline: SCHEDULED_POOL is a single thread shared by the whole bridge, and + // Agent.free() shuts down the StunStack, closes sockets and joins threads. + TaskPools.IO_POOL.submit { oldBundle.free() } + }, + transitionWindow.toMillis(), + TimeUnit.MILLISECONDS + ) + synchronized(restartLock) { + // Unless the bundle is already gone, in which case there is nothing left to free. + if (retiringBundle === oldBundle) { + transitionWindowTask = freeTask + } else { + freeTask.cancel(false) + } + } + } + + /** + * Gives up on an ICE restart whose new Agent never connected (it failed, or the timeout elapsed), keeping + * the established Agent in place — the transport itself does not fail. A no-op if the restart already cut + * over or was superseded. + */ + private fun abandonPendingRestart(bundle: AgentBundle, reason: String) { + synchronized(restartLock) { + if (pendingBundle !== bundle) { + return + } + pendingBundle = null + restartTimeoutTask?.cancel(false) + restartTimeoutTask = null + } + val elapsedMs = Duration.between(bundle.createdAt, clock.instant()).toMillis() + logger.warn( + "Abandoning ICE restart (generation=${bundle.generation}, local ufrag=${bundle.agent.localUfrag}) " + + "after ${elapsedMs}ms: $reason. Keeping the established Agent with local ufrag " + + "${currentBundle.agent.localUfrag}." + ) + iceRestartsFailed.inc() + // Not inline: this can run on the Agent's own state-change notification thread. + TaskPools.IO_POOL.submit { bundle.free() } + } + + private fun startReadingData(bundle: AgentBundle) { logger.cdebug { "Starting to read incoming data" } - val socket = iceComponent.selectedPair.iceSocketWrapper + val socket = bundle.component.selectedPair.iceSocketWrapper val receiveBuf = ByteArray(1500) val packet = DatagramPacket(receiveBuf, 0, receiveBuf.size) var receivedTime: Instant @@ -266,7 +757,9 @@ class IceTransport @JvmOverloads constructor( fun send(data: ByteArray, off: Int, length: Int) { if (running.get()) { try { - iceComponent.send(data, off, length) + // Always the established bundle: during a restart the new Agent is still running checks and we + // keep sending on the old selected pair until cutOver() swaps it in (make-before-break). + currentBundle.component.send(data, off, length) packetStats.numPacketsSent.increment() } catch (e: IOException) { logger.error("Error sending packet", e) @@ -280,9 +773,22 @@ class IceTransport @JvmOverloads constructor( fun stop() { if (running.compareAndSet(true, false)) { logger.info("Stopping") - iceAgent.removeStateChangeListener(iceStateChangeListener) - iceStream.removePairStateChangeListener(iceStreamPairChangedListener) - iceAgent.free() + val bundles = synchronized(restartLock) { + restartTimeoutTask?.cancel(false) + restartTimeoutTask = null + transitionWindowTask?.cancel(false) + transitionWindowTask = null + buildList { + add(currentBundle) + pendingBundle?.let { add(it) } + // A bundle inside its transition window would otherwise outlive the transport, until the + // task that frees it fires. + retiringBundle?.let { add(it) } + pendingBundle = null + retiringBundle = null + } + } + bundles.forEach { it.free() } } } @@ -294,6 +800,15 @@ class IceTransport @JvmOverloads constructor( put("iceWriteable", iceWriteable.get()) put("iceConnected", iceConnected.get()) put("iceFailed", iceFailed.get()) + put("localUfrag", currentBundle.agent.localUfrag) + put("iceGeneration", currentBundle.generation) + val pending = pendingBundle + put("restartPending", pending != null) + if (pending != null) { + put("pendingIceGeneration", pending.generation) + put("pendingLocalUfrag", pending.agent.localUfrag) + put("pendingChecksStarted", pending.checksStarted.get()) + } setAll(packetStats.toJson()) } @@ -301,10 +816,28 @@ class IceTransport @JvmOverloads constructor( if (!running.get()) { logger.warn("Not describing, transport is not running") } + // Prefer a pending restart's bundle: once we have rolled a new Agent, its credentials are the ones the + // peer must address its connectivity checks to (the USERNAME and MESSAGE-INTEGRITY of a check are + // built from the *peer's* view of our ufrag/password). Describing the old ones would send the peer + // checking against an Agent we are about to retire. With no restart in flight this is the established + // bundle, so the initial allocation path is unchanged. + val pending = pendingBundle + val bundle = pending ?: currentBundle with(pe) { - password = iceAgent.localPassword - ufrag = iceAgent.localUfrag - iceComponent.localCandidates?.forEach { cand -> + password = bundle.agent.localPassword + ufrag = bundle.agent.localUfrag + if (bundle === pending) { + // Stamp the generation of the restart round these credentials belong to. The peer echoes it + // back on the transport-info carrying its own new credentials, which is how we match its + // answer to this bundle and discard answers from a round we have since moved on from. + // + // Only for a pending bundle: the generation marks a transport the peer must restart against, + // and describing the established bundle (which keeps the generation it connected with) means + // the opposite — use what is already there. The peer rejects a generation that is not newer + // than the last one it saw, so stamping one here would have it drop the transport. + iceGeneration = bundle.generation + } + bundle.component.localCandidates?.forEach { cand -> cand.toCandidatePacketExtension(advertisePrivateAddresses)?.let { pe.addChildExtension(it) } } addChildExtension(IceRtcpmuxPacketExtension()) @@ -315,7 +848,11 @@ class IceTransport @JvmOverloads constructor( * @return the number of network reachable remote candidates contained in * the given list of candidates. */ - private fun addRemoteCandidates(remoteCandidates: List, iceAgentIsRunning: Boolean): Int { + private fun addRemoteCandidates( + bundle: AgentBundle, + remoteCandidates: List, + iceAgentIsRunning: Boolean + ): Int { var remoteCandidateCount = 0 // Sort the remote candidates (host < reflexive < relayed) in order to // create first the host, then the reflexive, the relayed candidates and @@ -324,14 +861,14 @@ class IceTransport @JvmOverloads constructor( remoteCandidates.sorted().forEach { candidate -> // Is the remote candidate from the current generation of the // iceAgent? - if (candidate.generation != iceAgent.generation) { + if (candidate.generation != bundle.agent.generation) { return@forEach } if (candidate.ipNeedsResolution() && !IceConfig.config.resolveRemoteCandidates) { logger.cdebug { "Ignoring remote candidate with non-literal address: ${candidate.ip}" } return@forEach } - val component = iceStream.getComponent(candidate.component) + val component = bundle.stream.getComponent(candidate.component) val remoteCandidate = RemoteCandidate( TransportAddress(candidate.ip, candidate.port, Transport.parse(candidate.protocol)), component, @@ -361,26 +898,41 @@ class IceTransport @JvmOverloads constructor( return remoteCandidateCount } - private fun iceStateChanged(ev: PropertyChangeEvent) { + private fun iceStateChanged(bundle: AgentBundle, ev: PropertyChangeEvent) { val oldState = ev.oldValue as IceProcessingState val newState = ev.newValue as IceProcessingState val transition = IceProcessingStateTransition(oldState, newState) - logger.debug("ICE state changed old=$oldState new=$newState") + val isPending = bundle === pendingBundle + val isCurrent = bundle === currentBundle + logger.debug( + "ICE state changed old=$oldState new=$newState for $bundle (pending=$isPending, current=$isCurrent)" + ) + + if (!isPending && !isCurrent) { + // A bundle we have already moved on from: either the pre-restart Agent inside its transition + // window, or one that was superseded and is being freed. It no longer speaks for the transport, so + // in particular it must not fail it. + logger.debug("Ignoring an ICE state change from a retired $bundle") + return + } when { transition.completed() -> { - if (iceConnected.compareAndSet(false, true)) { + if (isPending) { + // A restart's new Agent connected: swap it in and retire the old one. + cutOver(bundle) + } else if (iceConnected.compareAndSet(false, true)) { eventHandler?.connected() if (useUniquePort) { // ice4j's push API only works with the single port harvester. With unique ports we still need // to read from the socket. TaskPools.IO_POOL.submit { - startReadingData() + startReadingData(bundle) } } - if (iceComponent.selectedPair.remoteCandidate.type == CandidateType.RELAYED_CANDIDATE || - iceComponent.selectedPair.localCandidate.type == CandidateType.RELAYED_CANDIDATE + if (bundle.component.selectedPair.remoteCandidate.type == CandidateType.RELAYED_CANDIDATE || + bundle.component.selectedPair.localCandidate.type == CandidateType.RELAYED_CANDIDATE ) { iceSucceededRelayed.inc() } @@ -388,7 +940,11 @@ class IceTransport @JvmOverloads constructor( } } transition.failed() -> { - if (iceFailed.compareAndSet(false, true)) { + if (isPending) { + // Only the restart failed. The established Agent is untouched, so keep using it rather + // than failing the whole transport. + abandonPendingRestart(bundle, "the new Agent failed to connect") + } else if (iceFailed.compareAndSet(false, true)) { eventHandler?.failed() Companion.iceFailed.inc() } @@ -398,7 +954,7 @@ class IceTransport @JvmOverloads constructor( /** Update IceStatistics once an initial round-trip-time measurement is available. */ fun updateStatsOnInitialRtt(rttMs: Double) { - val selectedPair = iceComponent.selectedPair + val selectedPair = currentBundle.component.selectedPair val localCandidate = selectedPair?.localCandidate ?: return val harvesterName = if (localCandidate is HostCandidate) { "host" @@ -409,7 +965,13 @@ class IceTransport @JvmOverloads constructor( IceStatistics.stats.add(harvesterName, rttMs) } - private fun iceStreamPairChanged(ev: PropertyChangeEvent) { + private fun iceStreamPairChanged(bundle: AgentBundle, ev: PropertyChangeEvent) { + // Only the bundle we actually send on speaks for the connection's liveness. A lingering old bundle (in + // its transition window) or a pending one still running checks must not report writeability or refresh + // consent on the transport's behalf. + if (bundle !== currentBundle) { + return + } if (IceMediaStream.PROPERTY_PAIR_VALIDATED == ev.propertyName) { if (iceWriteable.compareAndSet(false, true)) { eventHandler?.writeable() @@ -453,6 +1015,72 @@ class IceTransport @JvmOverloads constructor( "ice_succeeded_relayed", "Number of times an ICE Agent succeeded and the selected pair included a relayed candidate." ) + + /** + * The total number of ICE restarts started (a new Agent was created for a peer-requested restart). + */ + val iceRestartsStarted = VideobridgeMetricsContainer.instance.registerCounter( + "ice_restarts_started", + "Number of ICE restarts started." + ) + + /** + * The total number of ICE restart requests this bridge could not honour at all: the feature is + * disabled, its timeout is misconfigured, or the transport is stopped. The endpoint is told to fall + * back to a full re-invite. + */ + val iceRestartsRejected = VideobridgeMetricsContainer.instance.registerCounter( + "ice_restarts_rejected", + "Number of ICE restart requests rejected because this bridge does not do ICE restarts (the " + + "feature is disabled or misconfigured, or the transport is stopped)." + ) + + /** + * The total number of ICE restart requests for a transport that had not connected yet, for which the + * existing Agent was kept. Ordinary traffic: the endpoint keeps a session that is still being + * negotiated, and nothing is lost. + */ + val iceRestartsNotEstablished = VideobridgeMetricsContainer.instance.registerCounter( + "ice_restarts_not_established", + "Number of ICE restart requests for a transport that was not established yet, for which the " + + "existing Agent was kept." + ) + + /** + * The total number of ICE restarts that could not create their new Agent, which means ice4j could not + * harvest or bind. Unlike the other two this indicates a problem with the bridge itself. + */ + val iceRestartsAgentCreationFailed = VideobridgeMetricsContainer.instance.registerCounter( + "ice_restarts_agent_creation_failed", + "Number of ICE restarts that failed because a new ICE Agent could not be created." + ) + + /** + * The total number of ICE restarts whose new Agent connected and was cut over to. + */ + val iceRestartsSucceeded = VideobridgeMetricsContainer.instance.registerCounter( + "ice_restarts_succeeded", + "Number of ICE restarts whose new Agent connected and was cut over to." + ) + + /** + * The total number of ICE restarts superseded by a newer one before they connected. Together with + * [iceRestartsSucceeded] and [iceRestartsFailed] this accounts for every restart in + * [iceRestartsStarted], except the ones still in flight. + */ + val iceRestartsSuperseded = VideobridgeMetricsContainer.instance.registerCounter( + "ice_restarts_superseded", + "Number of ICE restarts superseded by a newer restart before they connected." + ) + + /** + * The total number of ICE restarts abandoned because the new Agent failed or timed out. The + * established Agent is kept in these cases, so this does not imply the endpoint lost connectivity. + */ + val iceRestartsFailed = VideobridgeMetricsContainer.instance.registerCounter( + "ice_restarts_failed", + "Number of ICE restarts abandoned because the new Agent failed to connect." + ) } private class PacketStats { diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/websocket/ColibriWebSocketPassword.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/websocket/ColibriWebSocketPassword.kt new file mode 100644 index 0000000000..c27dc6234a --- /dev/null +++ b/jvb/src/main/kotlin/org/jitsi/videobridge/websocket/ColibriWebSocketPassword.kt @@ -0,0 +1,32 @@ +/* + * Copyright @ 2026 - Present, 8x8 Inc + * + * Licensed 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.jitsi.videobridge.websocket + +import java.security.SecureRandom + +private val random = SecureRandom() +private const val CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + +/** + * Generate a password which authenticates a colibri WebSocket. The password is used in a URL query parameter, so + * it only contains URL-safe characters. + * + * The peer receives the URL once, when we describe its transport, and re-dials that same URL each time the + * WebSocket reconnects. Because of this, the password must stay the same for the lifetime of the endpoint or + * relay which owns it. + */ +fun generateColibriWebSocketPassword(length: Int = 24) = + String(CharArray(length) { CHARS[random.nextInt(CHARS.length)] }) diff --git a/jvb/src/main/resources/reference.conf b/jvb/src/main/resources/reference.conf index f5ccddd5b4..6d0d29de07 100644 --- a/jvb/src/main/resources/reference.conf +++ b/jvb/src/main/resources/reference.conf @@ -315,6 +315,29 @@ videobridge { # addresses even for endpoints that have not signaled support for private addresses. # Note: Jicofo signals support for private addresses for jigasi and jibri. advertise-private-candidates = true + + # ICE restarts: when an endpoint explicitly requests one (colibri2 ), + # create a second ice4j Agent with freshly rotated local credentials and run it alongside the existing + # one. The old Agent keeps the selected pair and keeps sending until the new one connects + # (make-before-break), then we cut over to the new Agent. If disabled, such requests are rejected and + # the established Agent is left alone. + restart { + enabled = true + + # How long the old Agent is kept alive after we cut over to the new one. During this window both + # Agents keep their sockets, and ice4j routes each packet by the address it came from, so the + # endpoint's old-generation checks are answered by the old Agent rather than dropped. Set to 0 to + # free the old Agent immediately on cutover. + transition-window = 3 seconds + + # How long to wait for a new Agent to connect before giving up on the restart and keeping the + # existing Agent. Counted from when the Agent was created, so this covers the whole signaling + # round trip (bridge to jicofo to the endpoint and back), the endpoint gathering its own new + # candidates, and ICE itself -- on an endpoint whose network has just changed. A value that is + # not positive disables ICE restarts: the new Agent would be freed before the endpoint could + # answer it. + timeout = 20 seconds + } } transport { diff --git a/jvb/src/test/kotlin/org/jitsi/videobridge/transport/ice/IceTransportRestartTest.kt b/jvb/src/test/kotlin/org/jitsi/videobridge/transport/ice/IceTransportRestartTest.kt new file mode 100644 index 0000000000..642a40a870 --- /dev/null +++ b/jvb/src/test/kotlin/org/jitsi/videobridge/transport/ice/IceTransportRestartTest.kt @@ -0,0 +1,406 @@ +/* + * Copyright @ 2026 - Present, 8x8 Inc + * + * Licensed 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.jitsi.videobridge.transport.ice + +import io.kotest.core.spec.IsolationMode +import io.kotest.core.spec.style.ShouldSpec +import io.kotest.matchers.shouldBe +import io.mockk.every +import io.mockk.mockk +import io.mockk.verify +import org.ice4j.ice.Agent +import org.ice4j.ice.Component +import org.ice4j.ice.IceMediaStream +import org.ice4j.ice.IceProcessingState +import org.ice4j.ice.KeepAliveStrategy +import org.jitsi.config.withNewConfig +import org.jitsi.utils.concurrent.FakeScheduledExecutorService +import org.jitsi.utils.logging2.Logger +import org.jitsi.utils.logging2.LoggerImpl +import org.jitsi.videobridge.util.TaskPools +import org.jitsi.xmpp.extensions.jingle.IceUdpTransportPacketExtension +import java.beans.PropertyChangeEvent +import java.beans.PropertyChangeListener + +/** + * Tests the ICE restart state machine of [IceTransport]: which Agent is described, which one gets the peer's + * credentials, and which ones are freed. The ice4j [Agent]s are mocked, because real ones bind ports and can + * only change state by actually running ICE. + */ +class IceTransportRestartTest : ShouldSpec({ + isolationMode = IsolationMode.InstancePerLeaf + + // The restart timeout and the transition window are scheduled on TaskPools.SCHEDULED_POOL. + val scheduler = FakeScheduledExecutorService() + TaskPools.SCHEDULED_POOL = scheduler + afterSpec { TaskPools.resetScheduledPool() } + + val agents = FakeAgents() + fun createTransport() = IceTransport( + id = "test", + controlling = true, + useUniquePort = false, + advertisePrivateAddresses = false, + parentLogger = LoggerImpl("test"), + agentFactory = agents.factory + ) + + context("Before ICE is established") { + val transport = createTransport() + should("apply an ordinary transport update to the initial Agent") { + // The main negotiation path, which the restart routing sits in front of. + transport.startConnectivityEstablishment(remoteTransport(generation = null)) + + agents.created[0].remoteUfrag shouldBe "remote-ufrag" + agents.created[0].remotePassword shouldBe "remote-pwd" + agents.created[0].startCalls shouldBe 1 + } + should("keep the existing Agent instead of restarting") { + transport.requestIceRestart() shouldBe IceRestartResult.KEEP_EXISTING + agents.created.size shouldBe 1 + } + should("describe the existing Agent with no ice-generation") { + transport.requestIceRestart() + with(transport.describe()) { + ufrag shouldBe agents.created[0].ufrag + iceGeneration shouldBe IceUdpTransportPacketExtension.GENERATION_UNSPECIFIED + } + } + } + + context("With ICE established") { + val transport = createTransport() + val initial = agents.created[0] + initial.state = IceProcessingState.COMPLETED + + should("create a new Agent and describe it") { + transport.requestIceRestart() shouldBe IceRestartResult.STARTED + agents.created.size shouldBe 2 + with(transport.describe()) { + ufrag shouldBe agents.created[1].ufrag + password shouldBe agents.created[1].password + iceGeneration shouldBe 1 + } + } + should("keep sending on the established Agent until the new one connects") { + transport.requestIceRestart() + transport.send(ByteArray(10), 0, 10) + verify(exactly = 1) { initial.component.send(any(), any(), any()) } + verify(exactly = 0) { agents.created[1].component.send(any(), any(), any()) } + } + should("not start connectivity checks before the peer's credentials arrive") { + transport.requestIceRestart() + agents.created[1].startCalls shouldBe 0 + } + + context("A repeated request") { + should("re-describe the pending Agent while it has not started checks") { + transport.requestIceRestart() shouldBe IceRestartResult.STARTED + transport.requestIceRestart() shouldBe IceRestartResult.STARTED + + agents.created.size shouldBe 2 + agents.created[1].freed shouldBe false + transport.describe().iceGeneration shouldBe 1 + } + should("supersede a restart whose checks have started") { + transport.requestIceRestart() + transport.startConnectivityEstablishment(remoteTransport(generation = 1)) + agents.created[1].startCalls shouldBe 1 + + transport.requestIceRestart() shouldBe IceRestartResult.STARTED + agents.created.size shouldBe 3 + transport.describe().iceGeneration shouldBe 2 + awaitTrue { agents.created[1].freed } + } + } + + context("The peer's credentials") { + should("be applied when they carry the pending generation") { + transport.requestIceRestart() + transport.startConnectivityEstablishment(remoteTransport(generation = 1)) + + agents.created[1].remoteUfrag shouldBe "remote-ufrag" + agents.created[1].remotePassword shouldBe "remote-pwd" + agents.created[1].startCalls shouldBe 1 + } + should("be ignored when they carry another generation") { + transport.requestIceRestart() + transport.startConnectivityEstablishment(remoteTransport(generation = 7)) + + agents.created[1].remoteUfrag shouldBe null + agents.created[1].startCalls shouldBe 0 + } + should("not be taken from an untagged transport update") { + // The peer stamps the generation on the restart answer and on nothing else, so an untagged + // update is an ordinary one and carries the peer's old credentials. + transport.requestIceRestart() + transport.startConnectivityEstablishment(remoteTransport(generation = null)) + + agents.created[1].remoteUfrag shouldBe null + agents.created[1].startCalls shouldBe 0 + } + should("not start the checks twice") { + transport.requestIceRestart() + transport.startConnectivityEstablishment(remoteTransport(generation = 1)) + transport.startConnectivityEstablishment(remoteTransport(generation = 1)) + + agents.created[1].startCalls shouldBe 1 + } + should("not be replaced once the checks are running") { + // ice4j reads the remote credentials off the stream every time it signs a check or validates + // one, so a repeated update must not touch them. + transport.requestIceRestart() + transport.startConnectivityEstablishment(remoteTransport(generation = 1)) + transport.startConnectivityEstablishment( + remoteTransport(generation = 1, remoteUfrag = "other-ufrag", remotePassword = "other-pwd") + ) + + agents.created[1].remoteUfrag shouldBe "remote-ufrag" + agents.created[1].remotePassword shouldBe "remote-pwd" + } + } + + context("When the new Agent connects") { + should("cut over to it") { + transport.requestIceRestart() + agents.created[1].fireState(IceProcessingState.COMPLETED) + + transport.send(ByteArray(10), 0, 10) + verify(exactly = 1) { agents.created[1].component.send(any(), any(), any()) } + } + should("describe it without an ice-generation") { + // The generation marks a transport to restart against. The established one is not that. + transport.requestIceRestart() + agents.created[1].fireState(IceProcessingState.COMPLETED) + + with(transport.describe()) { + ufrag shouldBe agents.created[1].ufrag + iceGeneration shouldBe IceUdpTransportPacketExtension.GENERATION_UNSPECIFIED + } + } + should("free the old Agent only after the transition window") { + transport.requestIceRestart() + agents.created[1].fireState(IceProcessingState.COMPLETED) + + initial.freed shouldBe false + } + } + + context("When the new Agent fails") { + should("keep the established one") { + transport.requestIceRestart() + agents.created[1].fireState(IceProcessingState.FAILED) + + transport.hasFailed() shouldBe false + transport.describe().ufrag shouldBe initial.ufrag + awaitTrue { agents.created[1].freed } + } + should("not trigger the transport's own failure handling") { + var failed = false + transport.eventHandler = object : IceTransport.EventHandler { + override fun connected() {} + override fun failed() { + failed = true + } + override fun consentUpdated(time: java.time.Instant) {} + override fun writeable() {} + } + transport.requestIceRestart() + agents.created[1].fireState(IceProcessingState.FAILED) + + failed shouldBe false + } + } + + context("The scheduled tasks") { + should("free the old Agent when the transition window elapses") { + transport.requestIceRestart() + agents.created[1].fireState(IceProcessingState.COMPLETED) + initial.freed shouldBe false + + // Advances the fake clock to the task's deadline and runs it. + scheduler.runOne() + + awaitTrue { initial.freed } + } + should("abandon the restart when the timeout elapses") { + transport.requestIceRestart() + + scheduler.runOne() + + awaitTrue { agents.created[1].freed } + transport.hasFailed() shouldBe false + transport.describe().ufrag shouldBe initial.ufrag + } + should("not abandon a restart that has already cut over") { + transport.requestIceRestart() + agents.created[1].fireState(IceProcessingState.COMPLETED) + + // Runs the transition window task. The timeout task was cancelled by the cutover, so it is + // dropped rather than run, and does not free the Agent we just cut over to. + scheduler.runOne() + + awaitTrue { initial.freed } + agents.created[1].freed shouldBe false + transport.describe().ufrag shouldBe agents.created[1].ufrag + } + should("be cancelled by stop()") { + transport.requestIceRestart() + agents.created[1].fireState(IceProcessingState.COMPLETED) + transport.stop() + + scheduler.numPendingJobs() shouldBe 0 + } + } + + context("stop()") { + should("free the current and the pending Agent") { + transport.requestIceRestart() + transport.stop() + + initial.freed shouldBe true + agents.created[1].freed shouldBe true + } + should("free an Agent that is still inside its transition window") { + transport.requestIceRestart() + agents.created[1].fireState(IceProcessingState.COMPLETED) + transport.stop() + + initial.freed shouldBe true + agents.created[1].freed shouldBe true + } + } + + context("When the new Agent can not be created") { + should("report that a restart is unavailable") { + // A resource problem, so the endpoint is told to fall back to a full re-invite. + agents.failNext = true + transport.requestIceRestart() shouldBe IceRestartResult.UNAVAILABLE + transport.describe().ufrag shouldBe initial.ufrag + } + should("leave a later restart working") { + agents.failNext = true + transport.requestIceRestart() + transport.requestIceRestart() shouldBe IceRestartResult.STARTED + transport.describe().iceGeneration shouldBe 1 + } + } + } + + context("With ICE restarts disabled") { + should("report that a restart is unavailable") { + withNewConfig("videobridge.ice.restart.enabled = false") { + val transport = createTransport() + agents.created[0].state = IceProcessingState.COMPLETED + transport.requestIceRestart() shouldBe IceRestartResult.UNAVAILABLE + agents.created.size shouldBe 1 + } + } + } + + context("With a non-positive restart timeout") { + should("report that a restart is unavailable") { + withNewConfig("videobridge.ice.restart.timeout = 0 seconds") { + val transport = createTransport() + agents.created[0].state = IceProcessingState.COMPLETED + transport.requestIceRestart() shouldBe IceRestartResult.UNAVAILABLE + agents.created.size shouldBe 1 + } + } + } +}) + +private fun IceTransport.describe() = IceUdpTransportPacketExtension().also { describe(it) } + +private fun remoteTransport( + generation: Int?, + remoteUfrag: String = "remote-ufrag", + remotePassword: String = "remote-pwd" +) = IceUdpTransportPacketExtension().apply { + ufrag = remoteUfrag + password = remotePassword + generation?.let { iceGeneration = it } +} + +/** Waits for something another thread does (freeing an Agent happens on the IO pool). */ +private fun awaitTrue(timeoutMs: Long = 5000, condition: () -> Boolean) { + val deadline = System.currentTimeMillis() + timeoutMs + while (!condition() && System.currentTimeMillis() < deadline) { + Thread.sleep(10) + } + condition() shouldBe true +} + +private class FakeAgents { + val created = mutableListOf() + + /** Whether the next call to [factory] fails, the way a failure to bind a port would. */ + var failNext = false + + val factory: (Logger) -> Agent = { + if (failNext) { + failNext = false + throw java.io.IOException("Failed to bind") + } + FakeAgent(created.size).also { created.add(it) }.agent + } +} + +private class FakeAgent(index: Int) { + val ufrag = "ufrag-$index" + val password = "password-$index" + + var state: IceProcessingState = IceProcessingState.WAITING + + /** Written on the IO pool (that is where Agents are freed) and read from the test thread. */ + @Volatile + var freed = false + + var startCalls = 0 + var remoteUfrag: String? = null + var remotePassword: String? = null + + private val stateChangeListeners = mutableListOf() + + val component: Component = mockk(relaxed = true) + + val stream: IceMediaStream = mockk(relaxed = true) { + every { remoteUfrag = any() } answers { this@FakeAgent.remoteUfrag = firstArg() } + every { remoteUfrag } answers { this@FakeAgent.remoteUfrag } + every { remotePassword = any() } answers { this@FakeAgent.remotePassword = firstArg() } + every { remotePassword } answers { this@FakeAgent.remotePassword } + } + + val agent: Agent = mockk(relaxed = true) { + every { localUfrag } returns ufrag + every { localPassword } returns password + every { state } answers { this@FakeAgent.state } + every { createMediaStream(any()) } returns stream + every { + createComponent(any(), any(), any()) + } returns component + every { addStateChangeListener(any()) } answers { stateChangeListeners.add(firstArg()) } + every { removeStateChangeListener(any()) } answers { stateChangeListeners.remove(firstArg()) } + every { startConnectivityEstablishment() } answers { startCalls++ } + every { free() } answers { freed = true } + } + + fun fireState(newState: IceProcessingState, oldState: IceProcessingState = IceProcessingState.RUNNING) { + state = newState + val event = PropertyChangeEvent(agent, IceProcessingState::class.java.name, oldState, newState) + stateChangeListeners.toList().forEach { it.propertyChange(event) } + } +} diff --git a/pom.xml b/pom.xml index 4677fd0a70..79b212dcd8 100644 --- a/pom.xml +++ b/pom.xml @@ -149,7 +149,7 @@ ${project.groupId} jitsi-xmpp-extensions - 1.0-117-g60c0446 + 1.0-119-gc67bc81