CVE-2022-39368 (Java, published 2022-11-10T00:15Z)
Description: value: Eclipse Californium is a Java implementation of RFC7252 - Constrained Application Protocol for IoT Cloud services. In versions prior to 3.7.0, and 2.7.4, Californium is vulnerable to a Denial of Service. Failing handshakes don't cleanup counters for throttling, causing the threshold to be reached without being released again. This results in permanently dropping records. The issue was reported for certificate based handshakes, but may also affect PSK based handshakes. It generally affects client and server as well. This issue is patched in version 3.7.0 and 2.7.4. There are no known workarounds. main: commit 726bac57659410da463dcf404b3e79a7312ac0b9 2.7.x: commit 5648a0c27c2c2667c98419254557a14bac2b1f3f
Severity: nan (CVSS v3 base score: 8.2)
Weakness: CWE-459 \u2014 Incomplete Cleanup
The product does not properly "clean up" and remove temporary or supporting resources after they have been used.
Repository: https://github.com/eclipse-californium/californium
Files changed:
element-connector/src/main/java/org/eclipse/californium/elements/util/ExecutorsUtil.java
element-connector/src/main/java/org/eclipse/californium/elements/util/LimitedRunnable.java
element-connector/src/main/java/org/eclipse/californium/elements/util/SerialExecutor.java
scandium-core/api-changes.json
scandium-core/src/main/java/org/eclipse/californium/scandium/DTLSConnector.java
scandium-core/src/main/java/org/eclipse/californium/scandium/DtlsBindingPreprocessingConnector.java
scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/Connection.java
scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryConnectionStore.java
scandium-core/src/main/java/org/eclipse/californium/scandium/dtls/InMemoryReadWriteLockConnectionStore.java
scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorAdvancedTest.java
scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorHandshakeTest.java
scandium-core/src/test/java/org/eclipse/californium/scandium/DTLSConnectorTest.java
Commit message: Fix pending job counters.

Run pending jobs on connection shutdown to align the counters.

Fixes issue #2065

Signed-off-by: Achim Kraus <achim.kraus@cloudcoap.net>


Vulnerable code:
	 * Shutdown this executor and add all pending task from {@link #tasks} to
								setOwner();
								ExecutionListener current = listener.get();
										current.beforeExecution();
									command.run();
									LOGGER.error("unexpected error occurred:", t);
								} finally {
									try {
										if (current != null) {
											current.afterExecution();
										}
									} catch (Throwable t) {
										LOGGER.error("unexpected error occurred:", t);
									}
									clearOwner();
							} finally {
						} catch (RejectedExecutionException ex) {
							LOGGER.debug("shutdown?", ex);
	 * Called before and after executing a task.
						} else if (health instanceof DtlsHealthExtended) {
							((DtlsHealthExtended) health)
									.setConnections(maxConnections - connectionStore.remainingCapacity());
		for (Runnable job : pending) {
			try {
				job.run();
			} catch (Exception e) {
				LOGGER.warn("Shutdown DTLS connector:", e);
			}
		}
			executeInbound(getExecutorService(), peerAddress, new Runnable() {
					if (MDC_SUPPORT) {
						MDC.put("PEER", StringUtil.toString(firstRecord.getPeerAddress()));
					}
					processNewClientHello(firstRecord);
					if (MDC_SUPPORT) {
						MDC.clear();
					pendingInboundJobsCountdown.incrementAndGet();
				if (!executeInbound(serialExecutor, peerAddress, new Runnable() {
							if (running.get()) {
							pendingInboundJobsCountdown.incrementAndGet();
	protected boolean executeInbound(Executor executor, InetSocketAddress peer, Runnable job) {
		boolean pending = false;
			int count = pendingInboundJobsCountdown.decrementAndGet();
			if (count >= 0) {
				executor.execute(job);
				pending = true;
				if (health instanceof DtlsHealthExtended2) {
					((DtlsHealthExtended2) health).setPendingIncomingJobs(maxPendingInboundJobs - count);
				}
			} else {
			}
		} catch (RejectedExecutionException e) {
			// dont't terminate connection on shutdown!
			LOGGER.debug("Execution rejected while processing record from peer [{}]",
					StringUtil.toLog(peer), e);
		} finally {
			if (!pending) {
				if (health != null) {
					health.receivingRecord(true);
				}
				pendingInboundJobsCountdown.incrementAndGet();
		return pending;
		try {
			SerialExecutor executor = connection.getExecutor();
			if (!executeOutbound(executor, message.getInetSocketAddress(), new Runnable() {
				@Override
				public void run() {
					try {
						if (running.get()) {
							sendMessage(now, message, connection);
						} else {
							DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {}, connector not running!",
									message.getSize(), StringUtil.toLog(message.getInetSocketAddress()));
							message.onError(new InterruptedIOException("Connector is not running."));
							if (health != null) {
								health.sendingRecord(true);
							}
						}
					} catch (Exception e) {
						if (running.get()) {
							LOGGER.warn("Exception thrown by executor thread [{}]",
									Thread.currentThread().getName(), e);
						}
						DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {} {}", message.getSize(),
								StringUtil.toLog(message.getInetSocketAddress()), e.getMessage());
						message.onError(e);
					} finally {
						pendingOutboundJobsCountdown.incrementAndGet();
			})) {
				message.onError(new IllegalStateException("Outbound message overflow!"));
		} catch (RejectedExecutionException e) {
			LOGGER.debug("Execution rejected while sending application record [peer: {}]",
					StringUtil.toLog(message.getInetSocketAddress()), e);
			DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {}, {}", message.getSize(),
					StringUtil.toLog(message.getInetSocketAddress()), e.getMessage());
			message.onError(new InterruptedIOException("Connector is not running."));
	protected boolean executeOutbound(Executor executor, InetSocketAddress peer, Runnable job) {
		boolean pending = false;
			int count = pendingOutboundJobsCountdown.decrementAndGet();
			if (count >= 0) {
				executor.execute(job);
				pending = true;
				if (health instanceof DtlsHealthExtended2) {
					((DtlsHealthExtended2) health).setPendingOutgoingJobs(maxPendingOutboundJobs - count);
				}
			} else {
			}
		} finally {
			if (!pending) {
				if (health != null) {
					health.sendingRecord(true);
				}
				pendingOutboundJobsCountdown.incrementAndGet();
		return pending;
					executeHandshakeResult(serialExecutor, connection, new Runnable() {
								if (running.get()) {
								pendingHandshakeResultJobsCountdown.incrementAndGet();
				} catch (RejectedExecutionException e) {
					// dont't terminate connection on shutdown!
					LOGGER.debug("Execution rejected while processing handshake result [{}]", connection);
	protected boolean executeHandshakeResult(Executor executor, Connection connection, Runnable job) {
		boolean pending = false;
			int count = pendingHandshakeResultJobsCountdown.decrementAndGet();
			if (count >= 0) {
				executor.execute(job);
				pending = true;
				if (health instanceof DtlsHealthExtended2) {
					((DtlsHealthExtended2) health).setPendingHandshakeJobs(maxPendingHandshakeResultJobs - count);
				}
			} else {
			}
		} finally {
			if (!pending) {
				pendingHandshakeResultJobsCountdown.incrementAndGet();
		return pending;
	 * {@link DTLSConnector#executeInbound(Executor, InetSocketAddress, Runnable)}
	private SerialExecutor serialExecutor;
				if (connection.isExecuting()) {
					List<Runnable> pendings = connection.getExecutor().shutdownNow();
					if (LOGGER.isTraceEnabled()) {
						LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
								connections.size(), pendings.size(), new Throwable("connection removed!"));
					} else if (pendings.isEmpty()) {
						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					} else {
						LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
								connections.size(), pendings.size());
					}
					if (LOGGER.isTraceEnabled()) {
						LOGGER.trace("{}connection: remove {} (size {})", tag, connection, connections.size(),
								new Throwable("connection removed!"));
					} else {
						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					}
				if (connection.isExecuting()) {
					List<Runnable> pendings = connection.getExecutor().shutdownNow();
					if (LOGGER.isTraceEnabled()) {
						LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
								connections.size(), pendings.size(), new Throwable("connection removed!"));
					} else if (pendings.isEmpty()) {
						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					} else {
						LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
								connections.size(), pendings.size());
					}
					if (LOGGER.isTraceEnabled()) {
						LOGGER.trace("{}connection: remove {} (size {})", tag, connection, connections.size(),
								new Throwable("connection removed!"));
					} else {
						LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					}
	public void testDisableHelloVerifRequestForPsk() throws Exception {
	public void testDisabledHelloVerifRequestForPskWithCertificate() throws Exception {
	public void testDisabledHelloVerifRequestWithCertificate() throws Exception {
		connection.setConnectorContext(new SerialExecutor(executor), null);

import org.eclipse.californium.elements.util.SerialExecutor;
		connection2.setConnectorContext(new SerialExecutor(executor), null);
				.setConnectorContext(new SerialExecutor(executor), null)));
				.setConnectorContext(new SerialExecutor(executor), null)));
				.setConnectorContext(new SerialExecutor(executor), null)));

Fixed code:
	/**
	 * Run all jobs.
	 * 
	 * On {@link Throwable} write only warning and continue to run the other
	 * jobs.
	 * 
	 * @param jobs list of jobs to run.
	 * @since 3.7
	 */
	public static void runAll(List<Runnable> jobs) {
		for (Runnable job : jobs) {
			try {
				job.run();
			} catch (Throwable e) {
				LOGGER.warn("Ignoring error:", e);
			}
		}
	}

/*******************************************************************************
 * Copyright (c) 2022 Contributors to the Eclipse Foundation.
 * 
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v2.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 * 
 * The Eclipse Public License is available at
 *    http://www.eclipse.org/legal/epl-v20.html
 * and the Eclipse Distribution License is available at
 *    http://www.eclipse.org/org/documents/edl-v10.html.
 * 
 ******************************************************************************/
package org.eclipse.californium.elements.util;

import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Limited jobs.
 *
 * Limits pending jobs based on counters.
 * 
 * Note: for proper operation, ensure the at the end of
 * {@link Runnable#run()} the {@link LimitedRunnable#onDequeueing()} is
 * called.
 * 
 * <pre>
 *	public void run() {
 *		try {
 *			if (running.get() &amp;&amp; connection.isExecuting()) {
 *				// process
 *			}
 *		} finally {
 *			onDequeueing();
 *		}
 *	}
 * </pre>
 * 
 * @since 3.7
 */
public abstract class LimitedRunnable implements Runnable {

	/**
	 * Down counter to limit pending jobs.
	 */
	private final AtomicInteger counter;
	/**
	 * Indicator for overflows. {@code true}, if the counter indicates an
	 * overflow, {@code false}, otherwise.
	 */
	private volatile boolean overflow;

	/**
	 * Create job limited by the provided counter
	 * 
	 * @param counter counter in count-down mode
	 */
	public LimitedRunnable(AtomicInteger counter) {
		this.counter = counter;
	}

	/**
	 * Queue job.
	 * 
	 * @throws RejectedExecutionException if limit is exceeded
	 */
	public void onQueueing() {
		if (counter.decrementAndGet() < 0) {
			overflow = true;
			throw new RejectedExecutionException("queue overflow!");
		}
	}

	/**
	 * Dequeue job.
	 */
	public void onDequeueing() {
		counter.incrementAndGet();
	}

	/**
	 * Checks, if queueing this job causes an counter overflow.
	 * 
	 * @return {@code true}, if the counter indicates an overflow,
	 *         {@code false}, otherwise.
	 */
	public boolean isOverflown() {
		return overflow;
	}

	/**
	 * Execute this job.
	 * 
	 * @param executor executor to execute jobs.
	 */
	public void execute(Executor executor) {
		try {
			onQueueing();
			executor.execute(this);
		} catch (RejectedExecutionException ex) {
			onDequeueing();
			throw ex;
		}
	}
}
	 * @throws IllegalArgumentException if the executor is also a
	 *             {@link SerialExecutor}
		} else if (executor instanceof SerialExecutor) {
			throw new IllegalArgumentException("Sequences of SerialExecutors are not supported!");
	 * Shutdown this executor and add all pending jobs from {@link #tasks} to
							setOwner();
							ExecutionListener current = listener.get();
								if (current != null) {
									current.beforeExecution();
								}
								command.run();
							} catch (Throwable t) {
								LOGGER.error("unexpected error occurred:", t);
							} finally {
										current.afterExecution();
									LOGGER.error("unexpected error occurred after execution:", t);
								clearOwner();
							}
						} finally {
							try {
							} catch (RejectedExecutionException ex) {
								LOGGER.debug("shutdown?", ex);
	 * Called before and after executing a task. The calling thread is the same
	 * as the the one executing the job.
	],
	"3.7.0": [
		{
			"extension": "revapi.differences",
			"configuration": {
				"ignore": true,
				"differences": [
					{
						"code": "java.method.removed",
						"old": "method boolean org.eclipse.californium.scandium.DTLSConnector::executeHandshakeResult(java.util.concurrent.Executor, org.eclipse.californium.scandium.dtls.Connection, java.lang.Runnable)",
						"justification": "Not part of the public API"
					},
					{
						"code": "java.method.removed",
						"old": "method boolean org.eclipse.californium.scandium.DTLSConnector::executeInbound(java.util.concurrent.Executor, java.net.InetSocketAddress, java.lang.Runnable)",
						"justification": "Not part of the public API"
					},
					{
						"code": "java.method.removed",
						"old": "method boolean org.eclipse.californium.scandium.DTLSConnector::executeOutbound(java.util.concurrent.Executor, java.net.InetSocketAddress, java.lang.Runnable)",
						"justification": "Not part of the public API"
					}
				]
			}
		}
import org.eclipse.californium.elements.util.LimitedRunnable;
	/**
	 * Update health statistic.
	 * 
	 * Update {@link DtlsHealthExtended#setConnections(int)},
	 * {@link DtlsHealthExtended2#setPendingIncomingJobs(int)},
	 * {@link DtlsHealthExtended2#setPendingOutgoingJobs(int)}, and
	 * {@link DtlsHealthExtended2#setPendingHandshakeJobs(int)}.
	 * 
	 * @return {@code true}, if some pending jobs left, {@code false}, if not.
	 * @since 3.7
	 */
	public boolean updateHealth() {
		boolean pending = false;
		if (health instanceof DtlsHealthExtended) {
			((DtlsHealthExtended) health).setConnections(maxConnections - connectionStore.remainingCapacity());
		}
		if (health instanceof DtlsHealthExtended2) {
			DtlsHealthExtended2 health2 = (DtlsHealthExtended2) health;
			int jobs = maxPendingOutboundJobs - pendingOutboundJobsCountdown.get();
			health2.setPendingOutgoingJobs(jobs);
			if (jobs > 0) {
				LOGGER.debug("Pending out jobs {}", jobs);
			}
			pending = jobs > 0;
			jobs = maxPendingInboundJobs - pendingInboundJobsCountdown.get();
			health2.setPendingIncomingJobs(jobs);
			if (jobs > 0) {
				LOGGER.debug("Pending in jobs {}", jobs);
			}
			pending |= jobs > 0;
			jobs = maxPendingHandshakeResultJobs - pendingHandshakeResultJobsCountdown.get();
			health2.setPendingHandshakeJobs(jobs);
			if (jobs > 0) {
				LOGGER.debug("Pending handshake jobs {}", jobs);
			}
			pending |= jobs > 0;
		}
		return pending;
	}

						} else {
							updateHealth();
			updateHealth();
		ExecutorsUtil.runAll(pending);
			executeInbound(getExecutorService(), peerAddress, new LimitedRunnable(pendingInboundJobsCountdown) {
					try {
						if (running.get()) {
							if (MDC_SUPPORT) {
								MDC.put("PEER", StringUtil.toString(firstRecord.getPeerAddress()));
							}
							processNewClientHello(firstRecord);
							if (MDC_SUPPORT) {
								MDC.clear();
							}
						}
					} finally {
						onDequeueing();
				if (!executeInbound(serialExecutor, peerAddress, new LimitedRunnable(pendingInboundJobsCountdown) {
							if (running.get() && connection.isExecuting()) {
							onDequeueing();
	@NoPublicAPI
	protected boolean executeInbound(Executor executor, InetSocketAddress peer, LimitedRunnable job) {
			job.execute(executor);
			return true;
		} catch (RejectedExecutionException e) {
			if (job.isOverflown()) {
			} else {
				// dont't terminate connection on shutdown!
				LOGGER.debug("Execution rejected while processing record from peer [{}]", StringUtil.toLog(peer), e);
		if (health != null) {
			health.receivingRecord(true);
		}
		return false;
		SerialExecutor executor = connection.getExecutor();
		if (!executeOutbound(executor, message.getInetSocketAddress(), new LimitedRunnable(pendingOutboundJobsCountdown) {
			@Override
			public void run() {
				try {
					if (running.get() && connection.isExecuting()) {
						sendMessage(now, message, connection);
					} else {
						DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {}, connector not running!",
								message.getSize(), StringUtil.toLog(message.getInetSocketAddress()));
						message.onError(new InterruptedIOException("Connector is not running."));
				} catch (Exception e) {
					if (running.get()) {
						LOGGER.warn("Exception thrown by executor thread [{}]", Thread.currentThread().getName(), e);
					}
					DROP_LOGGER.trace("DTLSConnector drops {} outgoing bytes to {} {}", message.getSize(),
							StringUtil.toLog(message.getInetSocketAddress()), e.getMessage());
					if (health != null) {
						health.sendingRecord(true);
					}
					message.onError(e);
				} finally {
					onDequeueing();
		})) {
			message.onError(new IllegalStateException("Outbound message overflow!"));
	@NoPublicAPI
	protected boolean executeOutbound(Executor executor, InetSocketAddress peer, LimitedRunnable job) {
			job.execute(executor);
			return true;
		} catch (RejectedExecutionException e) {
			if (job.isOverflown()) {
			} else {
				// dont't terminate connection on shutdown!
				LOGGER.debug("Execution rejected while processing record to peer [{}]", StringUtil.toLog(peer), e);
		if (health != null) {
			health.sendingRecord(true);
		}
		return false;
					executeHandshakeResult(serialExecutor, connection, new LimitedRunnable(pendingHandshakeResultJobsCountdown) {
								if (running.get() && connection.isExecuting()) {
								onDequeueing();
	@NoPublicAPI
	protected boolean executeHandshakeResult(Executor executor, Connection connection, LimitedRunnable job) {
			job.execute(executor);
			return true;
		} catch (RejectedExecutionException e) {
			if (job.isOverflown()) {
			} else {
				// dont't terminate connection on shutdown!
				LOGGER.debug("Execution rejected while processing handshake result [{}]", connection, e);
		return false;
	 * {@link DTLSConnector#executeInbound(Executor, InetSocketAddress, org.eclipse.californium.scandium.DTLSConnector.LimitedRunnable)}
import java.util.List;
import org.eclipse.californium.elements.util.ExecutorsUtil;
	private volatile SerialExecutor serialExecutor;
	/**
	 * Shutdown executor and run all pending jobs.
	 * 
	 * The jobs are intended to check {@link #isExecuting()} in order to detect
	 * the shutdown.
	 * 
	 * @return number of pending jobs.
	 * @since 3.7
	 */
	public int shutdown() {
		SerialExecutor executor = getExecutor();
		if (executor != null) {
			List<Runnable> pendings = executor.shutdownNow();
			ExecutorsUtil.runAll(pendings);
			return pendings.size();
		}
		return 0;
	}

				int pendings = connection.shutdown();
				if (LOGGER.isTraceEnabled()) {
					LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
							connections.size(), pendings, new Throwable("connection removed!"));
				} else if (pendings == 0) {
					LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
							connections.size(), pendings);
				int pendings = connection.shutdown();
				if (LOGGER.isTraceEnabled()) {
					LOGGER.trace("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
							connections.size(), pendings, new Throwable("connection removed!"));
				} else if (pendings == 0) {
					LOGGER.debug("{}connection: remove {} (size {})", tag, connection, connections.size());
					LOGGER.debug("{}connection: remove {} (size {}, left jobs: {})", tag, connection,
							connections.size(), pendings);
	public void testDisableHelloVerifyRequestForPsk() throws Exception {
	public void testDisabledHelloVerifyRequestForPskWithCertificate() throws Exception {
	public void testDisabledHelloVerifyRequestWithCertificate() throws Exception {
		connection.setConnectorContext(executor, null);
import org.eclipse.californium.elements.util.TestCondition;
import org.eclipse.californium.elements.util.TestConditionTools;
	DtlsHealthLogger serverHealth;

		if (serverHelper != null && serverHelper.server != null) {
			assertThat(serverHelper.server.isRunning(), is(true));
			try {
				// wait until no pending jobs left
				TestConditionTools.waitForCondition(6000, 100, TimeUnit.MILLISECONDS, new TestCondition() {

					@Override
					public boolean isFulFilled() throws IllegalStateException {
						return !serverHelper.server.updateHealth();
					}
				});
				TestConditionTools.assertStatisticCounter("jobs left", serverHealth, "pending in jobs", is(0L));
				TestConditionTools.assertStatisticCounter("jobs left", serverHealth, "pending out jobs", is(0L));
				TestConditionTools.assertStatisticCounter("jobs left", serverHealth, "pending handshake jobs", is(0L));
			} catch (InterruptedException e) {
			}
		}
		serverHealth = new DtlsHealthLogger("server");
		serverBuilder.setHealthHandler(serverHealth);
		connection2.setConnectorContext(executor, null);
				.setConnectorContext(executor, null)));
				.setConnectorContext(executor, null)));
				.setConnectorContext(executor, null)));