Add Connect libp2p endpoint transport

README.md

@@ -13,12 +13,35 @@ low latency edge proxies network nearest to you.
 
 Please refer to https://connect.minekube.com for more documentation.
 
+## Connect libp2p endpoint mode
+
+The Connect libp2p endpoint path is enabled by configuring the Connect edge peer
+address. It keeps the normal WatchService path available as fallback while the
+endpoint also registers a stable libp2p peer for proxy-initiated session streams.
+
+Environment variables:
+
+- `CONNECT_LIBP2P_EDGE_ADDR`: comma-separated Connect edge libp2p multiaddrs,
+  each including `/p2p/<connect-edge-peer-id>`. Setting this enables libp2p mode.
+- `CONNECT_LIBP2P_LISTEN_ADDR`: optional endpoint listen multiaddrs. The
+  default is `/ip4/127.0.0.1/tcp/0`.
+- `CONNECT_LIBP2P_ADVERTISE_ADDRS`: optional explicit endpoint addresses
+  to publish instead of the local listen addresses.
+- `CONNECT_LIBP2P_RELAY_ADDRS`: optional relay bootstrap multiaddrs. The
+  endpoint reserves each relay through these addresses. During registration,
+  the Connect edge can challenge the endpoint to sign equivalent
+  `/p2p-circuit/p2p/<endpoint-peer-id>` addresses that are better for other
+  edge proxies to dial, such as private per-machine relay addresses.
+
 ## Working setups
 
 When installing the Connect plugin the following platform settings are supported.
 
 - PaperMC/Spigot
     - If running in Online mode you must set to `enforce-secure-profile: false` in [server.properties](https://minecraft.fandom.com/wiki/Server.properties)
+    - For Paper/Spigot endpoints, set `settings.connection-throttle: -1` in `bukkit.yml`.
+      Connect may retry the backend handshake while detecting the server's forwarding mode, and
+      Paper's default connection throttle can reject that retry as `Connection throttled!`.
     - ✔️️ No forwarding + Online mode
     - ✔️ No forwarding + Offline mode
     - ✔️ Velocity forwarding + Online/Offline mode

build-logic/build.gradle.kts

@@ -11,16 +11,14 @@ repositories {
 dependencies {
     implementation("net.kyori", "indra-common", "2.0.6")
     implementation("org.jfrog.buildinfo", "build-info-extractor-gradle", "4.26.1")
-    implementation("gradle.plugin.com.github.johnrengelman", "shadow", "7.1.1")
+    implementation("com.gradleup.shadow:shadow-gradle-plugin:8.3.11")
 }
 
 java {
-    sourceCompatibility = JavaVersion.VERSION_1_8
-    targetCompatibility = JavaVersion.VERSION_1_8
+    sourceCompatibility = JavaVersion.VERSION_11
+    targetCompatibility = JavaVersion.VERSION_11
 }
 
 tasks.withType<KotlinCompile> {
-    kotlinOptions {
-        jvmTarget = "1.8"
-    }
+    kotlinOptions.jvmTarget = "11"
 }

build-logic/src/main/kotlin/Versions.kt

@@ -37,6 +37,8 @@ object Versions {
     const val protocVersion = "3.19.4"
     const val bstatsVersion = "3.0.2"
     const val gsonVersion = "2.8.6"
+    const val jvmLibp2pVersion = "1.3.2-RELEASE"
+    const val kotlinStdlibVersion = "1.9.22"
 
     const val checkerQual = "3.19.0"
-}
+}

build-logic/src/main/kotlin/connect.base-conventions.gradle.kts

@@ -28,8 +28,8 @@ tasks {
 }
 
 java {
-    sourceCompatibility = JavaVersion.VERSION_1_8
-    targetCompatibility = JavaVersion.VERSION_1_8
+    sourceCompatibility = JavaVersion.VERSION_11
+    targetCompatibility = JavaVersion.VERSION_11
 
     withSourcesJar()
-}
+}

build-logic/src/main/kotlin/connect.shadow-conventions.gradle.kts

@@ -2,7 +2,7 @@ import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
 
 plugins {
     id("connect.base-conventions")
-    id("com.github.johnrengelman.shadow")
+    id("com.gradleup.shadow")
 }
 
 tasks {
@@ -51,4 +51,4 @@ fun callAddRelocations(configuration: Configuration, shadowJar: ShadowJar) =
     configuration.dependencies.forEach {
         if (it is ProjectDependency)
             addRelocations(it.dependencyProject, shadowJar)
-    }
+    }

build.gradle.kts

@@ -34,4 +34,4 @@ subprojects {
         in deployProjects -> plugins.apply("connect.shadow-conventions")
         else -> plugins.apply("connect.base-conventions")
     }
-}
+}

bungee/src/main/java/com/minekube/connect/BungeePlugin.java

@@ -31,6 +31,7 @@
 import com.minekube.connect.module.BungeeListenerModule;
 import com.minekube.connect.module.BungeePlatformModule;
 import com.minekube.connect.module.CommandModule;
+import com.minekube.connect.module.Libp2pEndpointModule;
 import com.minekube.connect.module.ProxyCommonModule;
 import com.minekube.connect.module.WatcherModule;
 import com.minekube.connect.util.ReflectionUtils;
@@ -64,6 +65,7 @@ public void onEnable() {
                 new CommandModule(),
                 new BungeeListenerModule(),
 //                new BungeeAddonModule(), - don't need proxy-side data injection
+                new Libp2pEndpointModule(),
                 new WatcherModule()
         );
     }

core/build.gradle.kts

@@ -16,12 +16,16 @@ dependencies {
     api("org.bstats", "bstats-base", Versions.bstatsVersion)
 
     implementation("com.squareup.okhttp3:okhttp:4.9.3")
+    implementation("io.libp2p:jvm-libp2p:${Versions.jvmLibp2pVersion}")
+    api("io.netty", "netty-transport", Versions.nettyVersion)
+    api("io.netty", "netty-codec", Versions.nettyVersion)
+    api("io.netty", "netty-transport-native-unix-common", Versions.nettyVersion)
+    implementation("org.jetbrains.kotlin:kotlin-stdlib:${Versions.kotlinStdlibVersion}")
     runtimeOnly("io.grpc", "grpc-netty-shaded", Versions.gRPCVersion)
     implementation("io.grpc", "grpc-protobuf", Versions.gRPCVersion)
     implementation("io.grpc", "grpc-stub", Versions.gRPCVersion)
     implementation("javax.annotation", "javax.annotation-api", "1.3.2")
 
-    // Test deps — pinned to versions still compatible with the Java 8 source target.
     testImplementation("org.junit.jupiter:junit-jupiter:5.10.5")
     testImplementation("org.mockito:mockito-core:4.11.0")
     testImplementation("org.awaitility:awaitility:4.2.2")
@@ -35,11 +39,6 @@ tasks.test {
     useJUnitPlatform()
 }
 
-// present on all platforms
-provided("io.netty", "netty-transport", Versions.nettyVersion)
-provided("io.netty", "netty-codec", Versions.nettyVersion)
-provided("io.netty", "netty-transport-native-unix-common", Versions.nettyVersion)
-
 relocate("org.bstats")
 
 configure<BlossomExtension> {

core/src/main/java/com/minekube/connect/ConnectPlatform.java

@@ -28,6 +28,7 @@
 import com.google.inject.Inject;
 import com.google.inject.Injector;
 import com.google.inject.Module;
+import com.google.inject.ConfigurationException;
 import com.google.inject.name.Named;
 import com.minekube.connect.api.ConnectApi;
 import com.minekube.connect.api.InstanceHolder;
@@ -42,6 +43,7 @@
 import com.minekube.connect.module.PostInitializeModule;
 import com.minekube.connect.register.WatcherRegister;
 import com.minekube.connect.tunnel.Tunneler;
+import com.minekube.connect.tunnel.p2p.Libp2pEndpoint;
 import com.minekube.connect.util.Metrics;
 import com.minekube.connect.util.UpdateChecker;
 import java.io.IOException;
@@ -135,6 +137,10 @@ public boolean enable(Module... postInitializeModules) {
     }
 
     public boolean disable() {
+        try {
+            guice.getInstance(Libp2pEndpoint.class).stop();
+        } catch (ConfigurationException ignored) {
+        }
         guice.getInstance(WatcherRegister.class).stop();
         guice.getInstance(Tunneler.class).close();
         guice.getInstance(CommonPlatformInjector.class).shutdown();

core/src/main/java/com/minekube/connect/module/CommonModule.java

@@ -32,6 +32,7 @@
 import com.google.inject.AbstractModule;
 import com.google.inject.Provides;
 import com.google.inject.Singleton;
+import com.google.inject.multibindings.Multibinder;
 import com.google.inject.name.Named;
 import com.minekube.connect.api.ConnectApi;
 import com.minekube.connect.api.SimpleConnectApi;
@@ -45,6 +46,9 @@
 import com.minekube.connect.inject.CommonPlatformInjector;
 import com.minekube.connect.packet.PacketHandlersImpl;
 import com.minekube.connect.platform.util.PlatformUtils;
+import com.minekube.connect.tunnel.TunnelClientTransport;
+import com.minekube.connect.tunnel.WebSocketTunnelTransport;
+import com.minekube.connect.tunnel.p2p.Libp2pTunnelTransport;
 import com.minekube.connect.util.Constants;
 import com.minekube.connect.util.HttpUtils;
 import com.minekube.connect.util.LanguageManager;
@@ -71,6 +75,10 @@ protected void configure() {
 
         bind(PacketHandlers.class).to(PacketHandlersImpl.class);
         bind(PacketHandlersImpl.class).asEagerSingleton();
+        Multibinder<TunnelClientTransport> transports =
+                Multibinder.newSetBinder(binder(), TunnelClientTransport.class);
+        transports.addBinding().to(WebSocketTunnelTransport.class);
+        transports.addBinding().to(Libp2pTunnelTransport.class);
     }
 
     @Provides
@@ -114,24 +122,29 @@ public OkHttpClient defaultOkHttpClient() {
 
     @Provides
     @Singleton
-    @Named("connectHttpClient")
-    public OkHttpClient connectOkHttpClient(
-            @Named("defaultHttpClient") OkHttpClient defaultOkHttpClient,
-            PlatformUtils platformUtils,
-            @Named("platformName") String implementationName,
-            ConnectApi api
-    ) throws IOException {
+    @Named("connectToken")
+    public String connectToken() throws IOException {
         Path tokenFile = dataDirectory.resolve("token.json");
 
         Optional<String> token = Token.load(tokenFile);
         if (!token.isPresent()) {
-            // Generate and save new token
             String t = Token.generate();
             Token.save(tokenFile, t);
             token = Optional.of(t);
         }
-        final String apiToken = token.get();
+        return token.get();
+    }
 
+    @Provides
+    @Singleton
+    @Named("connectHttpClient")
+    public OkHttpClient connectOkHttpClient(
+            @Named("defaultHttpClient") OkHttpClient defaultOkHttpClient,
+            PlatformUtils platformUtils,
+            @Named("platformName") String implementationName,
+            ConnectApi api,
+            @Named("connectToken") String apiToken
+    ) {
         return defaultOkHttpClient.newBuilder()
                 .addInterceptor(chain -> chain.proceed(chain.request().newBuilder()
                         // Add authorization token to every request

core/src/main/java/com/minekube/connect/module/Libp2pEndpointModule.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright (c) 2021-2022 Minekube. https://minekube.com
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ *
+ * @author Minekube
+ * @link https://github.com/minekube/connect-java
+ */
+
+package com.minekube.connect.module;
+
+import com.google.inject.AbstractModule;
+import com.minekube.connect.tunnel.p2p.Libp2pEndpoint;
+
+public final class Libp2pEndpointModule extends AbstractModule {
+
+    @Override
+    protected void configure() {
+        bind(Libp2pEndpoint.class).asEagerSingleton();
+    }
+}

core/src/main/java/com/minekube/connect/network/netty/LocalChannelInboundHandler.java

@@ -109,11 +109,18 @@ private String tunnelSvcAddr() {
     @Override
     public void channelActive(@NotNull ChannelHandlerContext ctx) throws Exception {
         // Start tunnel from downstream server -> upstream TunnelService
-        tunnelConn = tunneler.tunnel(
-                tunnelSvcAddr(),
-                context.getSessionProposal().getSession().getId(),
-                new TunnelHandler(logger, ctx.channel())
-        );
+        if (FORCE_TUNNEL_SERVICE_ADDR != null && !FORCE_TUNNEL_SERVICE_ADDR.isEmpty()) {
+            tunnelConn = tunneler.tunnel(
+                    tunnelSvcAddr(),
+                    context.getSessionProposal().getSession().getId(),
+                    new TunnelHandler(logger, ctx.channel())
+            );
+        } else {
+            tunnelConn = tunneler.tunnel(
+                    context.getSessionProposal().getSession(),
+                    new TunnelHandler(logger, ctx.channel())
+            );
+        }
         context.tunnelConn.set(tunnelConn);
         super.channelActive(ctx);
     }

core/src/main/java/com/minekube/connect/register/WatcherRegister.java

@@ -161,7 +161,8 @@ public void onOpen() {
 
         @Override
         public void onProposal(SessionProposal proposal) {
-            if (proposal.getSession().getTunnelServiceAddr().isEmpty()) {
+            if (proposal.getSession().getTunnelServiceAddr().isEmpty()
+                    && proposal.getSession().getTunnelTransportsCount() == 0) {
                 logger.info("Got session proposal with empty tunnel service address " +
                         "from WatchService, rejecting it");
                 proposal.reject(Status.newBuilder()
@@ -188,6 +189,7 @@ public void onProposal(SessionProposal proposal) {
                 return;
             }
 
+            tunneler.prepare(proposal.getSession());
             new LocalSession(logger, api, tunneler,
                     platformInjector.getServerSocketAddress(),
                     proposal