Add an experimental OpenZiti extension on a Netty client transport SPI

changes.md

@@ -8,7 +8,8 @@ Changes log
           Jetty 12. The connector class names are unchanged
           (`org.restlet.engine.connector.HttpServerHelper`, `HttpsServerHelper`,
           `HttpClientHelper`), and the Jetty dependencies are removed. No Netty type is exposed by
-          the public API: the module requires the Netty modules without `requires transitive`.
+          the public API, apart from the `NettyClientTransport` connector SPI: the module requires
+          the Netty modules without `requires transitive`.
         - Server calls run by default on virtual threads, one per call ("threadPool.virtual"
           parameter, `true` by default). Set it to `false` to use a pool of platform threads
           configured by the existing "threadPool.*" parameters.
@@ -283,7 +284,22 @@ Changes log
           default), the session token of temporary credentials and an unsigned payload option are
           given as parameters of the `ChallengeResponse`. The `HTTP_AWS_S3` scheme still uses
           Signature Version 2, which AWS no longer accepts for buckets created since June 2020.
+        - Added the experimental OpenZiti extension (`org.restlet.ext.ziti`, OpenZiti Java SDK
+          0.35.0): `ZitiClientTransport.install(client, zitiContext)` routes the calls of an HTTP
+          client to the origins an OpenZiti identity has a service for, such as
+          `crm.internal:443`, over the Ziti overlay network, without resolving their host name
+          with the system DNS, and the other calls over TCP. TLS, HTTP/1.1 and HTTP/2 run over it
+          as over TCP. Client side only. The transport listens to the context's events until
+          closed: close it once the client is stopped, or install another one, which closes it.
+        - Added the `NettyClientTransport` SPI to the Netty client connector, set with the
+          "org.restlet.engine.netty.clientTransport" client context attribute: it can route the
+          requests to some origins through another Netty channel and resolver than TCP, the HTTP
+          and SOCKS proxies and HTTP/3 being then bypassed, and the proxy credentials removed. It exposes Netty types and is meant for
+          extensions depending on Netty, such as the OpenZiti one.
     - Bugs fixed
+        - The Netty HTTP/2 client no longer turns a complete response into a communication error,
+          nor reports the request as disconnected, when the server answers before the end of the
+          request, such as with an immediate error, then closes the stream.
         - The Netty connectors now flush each full output buffer of a request or response body,
           so a producer that pauses after filling one no longer holds back what it produced
           until it resumes or ends. Issue #1632.

org.restlet.ext.ziti/pom.xml

@@ -0,0 +1,88 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.restlet</groupId>
+		<artifactId>org.restlet.parent</artifactId>
+		<version>3.0.0-SNAPSHOT</version>
+		<relativePath>../pom.xml</relativePath>
+	</parent>
+
+	<artifactId>org.restlet.ext.ziti</artifactId>
+	<name>Restlet Framework - OpenZiti extension</name>
+	<description>Routes HTTP client calls over an OpenZiti overlay network.</description>
+
+	<!-- The OpenZiti SDK pulls several versions of these: converge on the SDK's own Kotlin and on
+	     the Gson version of the Restlet build. -->
+	<dependencyManagement>
+		<dependencies>
+			<dependency>
+				<groupId>org.jetbrains.kotlin</groupId>
+				<artifactId>kotlin-stdlib</artifactId>
+				<version>2.4.20</version>
+			</dependency>
+			<dependency>
+				<groupId>org.jetbrains.kotlin</groupId>
+				<artifactId>kotlin-reflect</artifactId>
+				<version>2.4.20</version>
+			</dependency>
+			<dependency>
+				<groupId>org.jetbrains</groupId>
+				<artifactId>annotations</artifactId>
+				<version>23.0.0</version>
+			</dependency>
+			<dependency>
+				<groupId>com.google.code.gson</groupId>
+				<artifactId>gson</artifactId>
+				<version>${lib-gson-version}</version>
+			</dependency>
+		</dependencies>
+	</dependencyManagement>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.restlet</groupId>
+			<artifactId>org.restlet</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<!-- The OpenZiti SDK. Its ziti-netty channel isn't used, see ZitiSocketChannel. -->
+		<dependency>
+			<groupId>org.openziti</groupId>
+			<artifactId>ziti</artifactId>
+			<version>${lib-ziti-version}</version>
+			<exclusions>
+				<!-- Runtime dependencies of the SDK's own command line, which an application
+				     embedding it doesn't need: a logging backend and a CLI parser. -->
+				<exclusion>
+					<groupId>org.slf4j</groupId>
+					<artifactId>slf4j-simple</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>com.github.ajalt</groupId>
+					<artifactId>clikt</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<!-- The SDK's ServiceDetail model carries Jackson 2 annotations that the SDK only needs at
+		     run time; javac reads them when compiling against it. -->
+		<dependency>
+			<groupId>com.fasterxml.jackson.core</groupId>
+			<artifactId>jackson-annotations</artifactId>
+			<version>${lib-jackson-annotations-version}</version>
+			<scope>provided</scope>
+		</dependency>
+		<!-- Likewise for the kotlin.Metadata annotation of the SDK classes. -->
+		<dependency>
+			<groupId>org.jetbrains.kotlin</groupId>
+			<artifactId>kotlin-stdlib</artifactId>
+			<scope>provided</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-api</artifactId>
+			<version>${lib-junit-version}</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+</project>

org.restlet.ext.ziti/src/main/java/module-info.java

@@ -0,0 +1,21 @@
+/**
+ * Copyright 2005-2026 Qlik
+ *
+ * <p>The content of this file is subject to the terms of the Apache 2.0 open source license
+ * available at https://www.opensource.org/licenses/apache-2.0
+ *
+ * <p>Restlet is a registered trademark of QlikTech International AB.
+ */
+@SuppressWarnings({"requires-automatic", "requires-transitive-automatic"})
+module org.restlet.ext.ziti {
+    requires transitive org.restlet;
+    // The OpenZiti SDK jars are not modular and declare no Automatic-Module-Name.
+    requires transitive ziti;
+    requires edge.api;
+    requires io.netty.buffer;
+    requires io.netty.common;
+    requires io.netty.transport;
+    requires io.netty.resolver;
+
+    exports org.restlet.ext.ziti;
+}

org.restlet.ext.ziti/src/main/java/org/restlet/ext/ziti/ZitiClientTransport.java

@@ -0,0 +1,255 @@
+/**
+ * Copyright 2005-2026 Qlik
+ *<p>
+ * The content of this file is subject to the terms of the Apache 2.0 open
+ * source license available at https://www.opensource.org/licenses/apache-2.0
+ *<p>
+ * Restlet is a registered trademark of QlikTech International AB.
+ */
+package org.restlet.ext.ziti;
+
+import io.netty.resolver.NoopAddressResolverGroup;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.openziti.ZitiContext;
+import org.restlet.Client;
+import org.restlet.Context;
+import org.restlet.engine.netty.NettyClientTransport;
+
+/**
+ * Routes the HTTP client calls to the origins an OpenZiti identity can reach over the Ziti overlay
+ * network, and the others over TCP.
+ *
+ * <p>An origin is routed over Ziti when the context has a service intercepting its host and port
+ * ({@link ZitiContext#getService(InetSocketAddress, long)}), such as {@code crm.internal:443}. Its
+ * host name is then never resolved by the system DNS: the connection is a {@link ZitiSocketChannel}
+ * that dials that service. Every other origin is reached over TCP.
+ *
+ * <p>The context loads its services after authenticating. Until it has (its status is {@link
+ * ZitiContext.Status.Active}), routing waits, up to a timeout counted from the creation of the
+ * transport (10 seconds by default). Past it, the origins are reached over TCP until the services
+ * are loaded: create the transport early, or wait for the context to be active, so that no
+ * intercepted host name goes to the DNS. The lookups of the origins are cached, and the cache is
+ * cleared on each service event of the context. TLS, HTTP/1.1 and HTTP/2 run over it like over TCP:
+ * an {@code https} origin is encrypted end to end on top of the overlay's own encryption.
+ *
+ * <pre>
+ * ZitiContext ztx = Ziti.newContext(identityFile, new char[0]);
+ * Client client = new Client(Protocol.HTTPS);
+ * ZitiClientTransport.install(client, ztx);
+ * </pre>
+ *
+ * <p>The context is not owned: destroy it after the client is stopped. The transport listens to the
+ * events of the context: close it once the client is stopped, to stop listening while the context
+ * lives on. Installing a new transport on a client closes the one it replaces.
+ */
+public class ZitiClientTransport implements NettyClientTransport, AutoCloseable {
+
+    private static final Logger LOGGER = Logger.getLogger(ZitiClientTransport.class.getName());
+
+    /** The default time to wait for the services of the context. */
+    private static final Duration DEFAULT_SERVICES_TIMEOUT = Duration.ofSeconds(10);
+
+    /** The maximum number of origins whose lookup is cached. */
+    private static final int MAX_CACHED_ORIGINS = 1024;
+
+    /** The Ziti context. */
+    private final ZitiContext context;
+
+    /** The route over Ziti, shared by every intercepted origin. */
+    private final Route route;
+
+    /** Completed once the context has loaded its services. */
+    private final CompletableFuture<Void> servicesLoaded = new CompletableFuture<>();
+
+    /** The time, in {@link System#nanoTime()}, past which routing no longer waits for them. */
+    private final long servicesDeadline;
+
+    /** Whether each origin, as host:port, is intercepted by a service. */
+    private final Map<String, Boolean> intercepted = new ConcurrentHashMap<>();
+
+    /** Counts the service events, so that a lookup made meanwhile isn't cached. */
+    private final AtomicLong serviceEvents = new AtomicLong();
+
+    /** The subscription to the status of the context. */
+    private final Future<?> statusSubscription;
+
+    /** The subscription to the service events of the context. */
+    private final Future<?> serviceSubscription;
+
+    /** Whether the transport is closed: it then no longer caches the lookups. */
+    private volatile boolean closed;
+
+    /**
+     * Constructor, waiting up to 10 seconds for the services of the context.
+     *
+     * @param context The Ziti context of the identity used to dial the services.
+     */
+    public ZitiClientTransport(ZitiContext context) {
+        this(context, DEFAULT_SERVICES_TIMEOUT);
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param context The Ziti context of the identity used to dial the services.
+     * @param servicesTimeout The time to wait for the context to load its services, from now.
+     */
+    public ZitiClientTransport(ZitiContext context, Duration servicesTimeout) {
+        this.context = context;
+        // The channel maps the host and port to the Ziti service itself.
+        this.route =
+                new Route(() -> new ZitiSocketChannel(context), NoopAddressResolverGroup.INSTANCE);
+        this.servicesDeadline = System.nanoTime() + servicesTimeout.toNanos();
+        // Both run until closed. The status is active once the services are loaded.
+        this.statusSubscription =
+                context.onStatus(
+                        status -> {
+                            if (status instanceof ZitiContext.Status.Active) {
+                                servicesLoaded.complete(null);
+                            }
+                        });
+        this.serviceSubscription =
+                context.onServiceEvent(
+                        event -> {
+                            serviceEvents.incrementAndGet();
+                            intercepted.clear();
+                        });
+    }
+
+    /**
+     * Routes a client's calls to the origins the Ziti identity can reach over Ziti. A Ziti
+     * transport installed before on the client is closed.
+     *
+     * @param client The client, before it is started. A context is set on it if it has none.
+     * @param context The Ziti context of the identity used to dial the services.
+     * @return The transport set on the client.
+     */
+    public static ZitiClientTransport install(Client client, ZitiContext context) {
+        ZitiClientTransport transport = new ZitiClientTransport(context);
+        if (client.getContext() == null) {
+            client.setContext(new Context());
+        }
+        Object replaced =
+                client.getContext().getAttributes().put(NettyClientTransport.ATTRIBUTE, transport);
+        if (replaced instanceof ZitiClientTransport previous) {
+            previous.close();
+        }
+        return transport;
+    }
+
+    /**
+     * Stops listening to the events of the context, without destroying it. A closed transport still
+     * routes, without caching the lookups since it no longer sees the service events.
+     */
+    @Override
+    public void close() {
+        closed = true;
+        statusSubscription.cancel(true);
+        serviceSubscription.cancel(true);
+        intercepted.clear();
+    }
+
+    /**
+     * Returns the Ziti context.
+     *
+     * @return The Ziti context.
+     */
+    public ZitiContext getContext() {
+        return context;
+    }
+
+    @Override
+    public Route route(String host, int port) {
+        if (!awaitServices()) {
+            return null;
+        }
+
+        String origin = host + ":" + port;
+        Boolean routed = intercepted.get(origin);
+        if (routed == null) {
+            long events = serviceEvents.get();
+            routed = isIntercepted(host, port);
+            if (!closed
+                    && (serviceEvents.get() == events)
+                    && (intercepted.size() < MAX_CACHED_ORIGINS)) {
+                intercepted.put(origin, routed);
+                // A service event counted after the check above may have cleared the cache
+                // before the put: the result may be stale, so it is uncached. One counted after
+                // this check clears it afterwards.
+                if (serviceEvents.get() != events) {
+                    intercepted.remove(origin, routed);
+                }
+            }
+        }
+        return routed ? route : null;
+    }
+
+    /**
+     * Waits for the context to load its services, until the deadline.
+     *
+     * @return True if they are loaded.
+     */
+    private boolean awaitServices() {
+        if (servicesLoaded.isDone()) {
+            return true;
+        }
+
+        long left = servicesDeadline - System.nanoTime();
+        try {
+            if (left > 0) {
+                servicesLoaded.get(left, TimeUnit.NANOSECONDS);
+                return true;
+            }
+        } catch (InterruptedException _) {
+            Thread.currentThread().interrupt();
+        } catch (TimeoutException | ExecutionException _) {
+            // Not loaded in time.
+        }
+
+        LOGGER.log(
+                Level.WARNING,
+                () ->
+                        "The Ziti context "
+                                + context.name()
+                                + " has not loaded its services: connecting over TCP");
+        return false;
+    }
+
+    /**
+     * Indicates whether a service of the context intercepts an origin, without waiting for a
+     * service event: {@link ZitiContext#getService(InetSocketAddress)} waits for one, forever if
+     * the identity has no service.
+     *
+     * @param host The host name of the origin.
+     * @param port The port of the origin.
+     * @return True if a service intercepts it.
+     */
+    private boolean isIntercepted(String host, int port) {
+        // Looks the loaded services up, then times out at once. Kotlin doesn't declare the
+        // TimeoutException thrown when no service intercepts it: the lookup declares it.
+        ServiceLookup lookup =
+                () -> context.getService(InetSocketAddress.createUnresolved(host, port), 0L);
+        try {
+            return lookup.get() != null;
+        } catch (TimeoutException _) {
+            return false;
+        }
+    }
+
+    /** A service lookup declaring the TimeoutException that the SDK throws undeclared. */
+    @FunctionalInterface
+    private interface ServiceLookup {
+        Object get() throws TimeoutException;
+    }
+}