working?

build.gradle.kts

@@ -24,6 +24,8 @@ repositories {
 
 dependencies {
     implementation("us.myles:viaversion:3.1.0")
+    implementation("nl.matsv:viabackwards-all:3.1.0")
+    implementation("de.gerrygames:viarewind-all:1.5.1")
     implementation("io.netty:netty-all:4.1.51.Final")
     implementation(kotlin("stdlib-jdk8"))
 }

src/main/kotlin/com/github/creeper123123321/viaaas/CloudCodec.kt

@@ -1,13 +1,13 @@
 package com.github.creeper123123321.viaaas
 
 import io.netty.buffer.ByteBuf
+import io.netty.buffer.Unpooled
 import io.netty.channel.Channel
 import io.netty.channel.ChannelHandler
 import io.netty.channel.ChannelHandlerContext
 import io.netty.channel.ChannelInitializer
-import io.netty.handler.codec.MessageToMessageDecoder
-import io.netty.handler.codec.MessageToMessageEncoder
-import io.netty.handler.codec.ReplayingDecoder
+import io.netty.handler.codec.*
+import io.netty.handler.flow.FlowControlHandler
 import io.netty.handler.timeout.ReadTimeoutHandler
 import us.myles.ViaVersion.api.data.UserConnection
 import us.myles.ViaVersion.api.protocol.ProtocolPipeline
@@ -15,12 +15,106 @@ import us.myles.ViaVersion.api.type.Type
 import us.myles.ViaVersion.exception.CancelCodecException
 import us.myles.ViaVersion.exception.CancelDecoderException
 import us.myles.ViaVersion.exception.CancelEncoderException
-import us.myles.ViaVersion.packets.State
 import us.myles.ViaVersion.util.PipelineUtil
 import java.util.concurrent.TimeUnit
+import java.util.zip.Deflater
+import java.util.zip.Inflater
+
+
+object ChannelInit : ChannelInitializer<Channel>() {
+    override fun initChannel(ch: Channel) {
+        val user = UserConnection(ch)
+        ProtocolPipeline(user).add(CloudHandlerProtocol())
+        ch.pipeline().addLast("frame-encoder", FrameEncoder)
+                .addLast("frame-decoder", FrameDecoder())
+                .addLast("compress", CloudCompressor())
+                .addLast("decompress", CloudDecompressor())
+                .addLast("via-encoder", CloudEncodeHandler(user))
+                .addLast("via-decoder", CloudDecodeHandler(user))
+                .addLast("flow-handler", FlowControlHandler())
+                .addLast("timeout", ReadTimeoutHandler(30, TimeUnit.SECONDS))
+                .addLast("handler", CloudSideForwarder(user, null))
+    }
+}
+
+class BackendInit(val user: UserConnection) : ChannelInitializer<Channel>() {
+    override fun initChannel(ch: Channel) {
+        ch.pipeline().addLast("frame-encoder", FrameEncoder)
+                .addLast("frame-decoder", FrameDecoder())
+                .addLast("compress", CloudCompressor())
+                .addLast("decompress", CloudDecompressor())
+                .addLast("timeout", ReadTimeoutHandler(30, TimeUnit.SECONDS))
+                .addLast("handler", CloudSideForwarder(user, null))
+    }
+}
+
+class CloudDecompressor(var threshold: Int = -1) : MessageToMessageDecoder<ByteBuf>() {
+    // https://github.com/Gerrygames/ClientViaVersion/blob/master/src/main/java/de/gerrygames/the5zig/clientviaversion/netty/CompressionEncoder.java
+    private val inflater: Inflater = Inflater()
+
+    @Throws(Exception::class)
+    override fun decode(ctx: ChannelHandlerContext, input: ByteBuf, out: MutableList<Any>) {
+        if (threshold == -1) {
+            out.add(input.retain())
+            return
+        }
+        if (input.readableBytes() != 0) {
+            val outLength = Type.VAR_INT.readPrimitive(input)
+            if (outLength == 0) {
+                out.add(input.readBytes(input.readableBytes()))
+            } else {
+                if (outLength < threshold) {
+                    throw DecoderException("Badly compressed packet - size of $outLength is below server threshold of $threshold")
+                }
+                if (outLength > 2097152) {
+                    throw DecoderException("Badly compressed packet - size of $outLength is larger than protocol maximum of 2097152")
+                }
+                val temp = ByteArray(input.readableBytes())
+                input.readBytes(temp)
+                inflater.setInput(temp)
+                val output = ByteArray(outLength)
+                inflater.inflate(output)
+                out.add(Unpooled.wrappedBuffer(output))
+                inflater.reset()
+            }
+        }
+    }
+
+}
+
+class CloudCompressor(var threshold: Int = -1) : MessageToByteEncoder<ByteBuf>() {
+    // https://github.com/Gerrygames/ClientViaVersion/blob/master/src/main/java/de/gerrygames/the5zig/clientviaversion/netty/CompressionEncoder.java
+    private val buffer = ByteArray(8192)
+    private val deflater: Deflater = Deflater()
+
+    @Throws(Exception::class)
+    override fun encode(ctx: ChannelHandlerContext, input: ByteBuf, out: ByteBuf) {
+        if (threshold == -1) {
+            out.writeBytes(input)
+            return
+        }
+        val frameLength = input.readableBytes()
+        if (frameLength < threshold) {
+            Type.VAR_INT.writePrimitive(out, 0)
+            out.writeBytes(input)
+        } else {
+            Type.VAR_INT.writePrimitive(out, frameLength)
+            val inBytes = ByteArray(frameLength)
+            input.readBytes(inBytes)
+            deflater.setInput(inBytes, 0, frameLength)
+            deflater.finish()
+            while (!deflater.finished()) {
+                val written = deflater.deflate(buffer)
+                out.writeBytes(buffer, 0, written)
+            }
+            deflater.reset()
+        }
+    }
+
+}
 
 @ChannelHandler.Sharable
-object FrameCoder : MessageToMessageEncoder<ByteBuf>() {
+object FrameEncoder : MessageToMessageEncoder<ByteBuf>() {
     override fun encode(ctx: ChannelHandlerContext, msg: ByteBuf, out: MutableList<Any>) {
         val length = ctx.alloc().buffer(5)
         Type.VAR_INT.writePrimitive(length, msg.readableBytes())
@@ -32,22 +126,9 @@ object FrameCoder : MessageToMessageEncoder<ByteBuf>() {
 class FrameDecoder : ReplayingDecoder<ByteBuf>() {
     override fun decode(ctx: ChannelHandlerContext, input: ByteBuf, out: MutableList<Any>) {
         val length = Type.VAR_INT.readPrimitive(input)
-        if (length >= 2097152) throw IndexOutOfBoundsException()
+        if (length >= 2097152 || length < 0) throw DecoderException("Invalid length!")
         out.add(input.readRetainedSlice(length))
-    }
-}
-
-object ChannelInit : ChannelInitializer<Channel>() {
-    override fun initChannel(ch: Channel) {
-        ch.pipeline().addLast("frame-encoder", FrameCoder)
-                .addLast("frame-decoder", FrameDecoder())
-                .addLast("timeout", ReadTimeoutHandler(30, TimeUnit.SECONDS))
-                .addLast("handler", CloudSideForwarder(null))
-        val user = UserConnection(ch)
-        ProtocolPipeline(user)
-
-        ch.pipeline().addBefore("encoder", "via-encoder", CloudEncodeHandler(user))
-        ch.pipeline().addBefore("decoder", "via-decoder", CloudDecodeHandler(user))
+        checkpoint()
     }
 }
 
@@ -60,18 +141,6 @@ class CloudDecodeHandler(val info: UserConnection) : MessageToMessageDecoder<Byt
         }
         val transformedBuf: ByteBuf = ctx.alloc().buffer().writeBytes(bytebuf)
         try {
-            if (info.protocolInfo!!.state == State.HANDSHAKE) {
-                val id = Type.VAR_INT.readPrimitive(transformedBuf)
-                Type.VAR_INT.writePrimitive(transformedBuf, id)
-                if (id == 0 && info.get(CloudData::class.java) == null) {
-                    val ver = Type.VAR_INT.readPrimitive(transformedBuf) // Client ver
-                    Type.VAR_INT.writePrimitive(transformedBuf, ver)
-                    val origAddr = Type.STRING.read(transformedBuf)
-                    val addr = origAddr.split(" ")[0].split(".")
-                    val port = Type.SHORT.readPrimitive(transformedBuf)
-                }
-            }
-
             info.transformIncoming(transformedBuf, CancelDecoderException::generate)
             out.add(transformedBuf.retain())
         } finally {

src/main/kotlin/com/github/creeper123123321/viaaas/CloudHandler.kt

@@ -1,17 +1,73 @@
 package com.github.creeper123123321.viaaas
 
+import com.google.gson.Gson
 import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufAllocator
 import io.netty.channel.Channel
 import io.netty.channel.ChannelHandlerContext
 import io.netty.channel.SimpleChannelInboundHandler
-import io.netty.util.ReferenceCountUtil
+import us.myles.ViaVersion.api.data.UserConnection
+import us.myles.ViaVersion.api.type.Type
+import us.myles.ViaVersion.packets.State
+
+class CloudSideForwarder(val userConnection: UserConnection, var other: Channel?) : SimpleChannelInboundHandler<ByteBuf>() {
 
-class CloudSideForwarder(var other: Channel?): SimpleChannelInboundHandler<ByteBuf>() {
     override fun channelRead0(ctx: ChannelHandlerContext, msg: ByteBuf) {
-        try {
-            other?.write(msg)
-        } finally {
-            ReferenceCountUtil.release(msg)
+        other?.write(msg.retain())
+    }
+
+    override fun channelInactive(ctx: ChannelHandlerContext) {
+        super.channelInactive(ctx)
+        println(userConnection.channel?.remoteAddress().toString() + " was disconnected")
+        other?.close()
+    }
+
+    override fun channelReadComplete(ctx: ChannelHandlerContext?) {
+        super.channelReadComplete(ctx)
+        other?.flush()
+    }
+
+    override fun channelWritabilityChanged(ctx: ChannelHandlerContext) {
+        super.channelWritabilityChanged(ctx)
+        other?.setAutoRead(ctx.channel().isWritable)
+    }
+
+    override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
+        disconnect("Exception: $cause")
+        cause.printStackTrace()
+    }
+
+    fun disconnect(s: String) {
+        if (userConnection.channel?.isActive != true) return
+
+        val msg = "[VIAaaS] $s";
+        println("Disconnecting " + userConnection.channel!!.remoteAddress() + ": " + s)
+        when (userConnection.protocolInfo!!.state) {
+            State.LOGIN -> {
+                val packet = ByteBufAllocator.DEFAULT.buffer()
+                try {
+                    packet.writeByte(0) // id 0 disconnect
+                    Type.STRING.write(packet, Gson().toJson(msg))
+                    userConnection.sendRawPacketFuture(packet.retain()).addListener { userConnection.channel?.close() }
+                } finally {
+                    packet.release()
+                }
+            }
+            State.STATUS -> {
+                val packet = ByteBufAllocator.DEFAULT.buffer()
+                try {
+                    packet.writeByte(0) // id 0 disconnect
+                    Type.STRING.write(packet, """{"version": {"name": "VIAaaS","protocol": -1},
+"players": {"max": 0,"online": 0,"sample": []},	
+"description": {"text": ${Gson().toJson(msg)}}}""")
+                    userConnection.sendRawPacketFuture(packet.retain()).addListener { userConnection.channel?.close() }
+                } finally {
+                    packet.release()
+                }
+            }
+            else -> {
+                userConnection.disconnect(s)
+            }
         }
     }
 }

src/main/kotlin/com/github/creeper123123321/viaaas/CloudPlatform.kt

@@ -2,6 +2,7 @@ package com.github.creeper123123321.viaaas
 
 import io.netty.buffer.ByteBuf
 import io.netty.channel.DefaultEventLoop
+import io.netty.channel.socket.SocketChannel
 import us.myles.ViaVersion.AbstractViaConfig
 import us.myles.ViaVersion.api.Via
 import us.myles.ViaVersion.api.ViaAPI
@@ -66,6 +67,7 @@ object CloudAPI : ViaAPI<Unit> {
     override fun sendRawPacket(p0: Unit?, p1: ByteBuf?) {
         TODO("Not yet implemented")
     }
+
     override fun sendRawPacket(p0: UUID?, p1: ByteBuf?) {
         TODO("Not yet implemented")
     }
@@ -73,9 +75,11 @@ object CloudAPI : ViaAPI<Unit> {
     override fun getPlayerVersion(p0: Unit?): Int {
         TODO("Not yet implemented")
     }
+
     override fun getPlayerVersion(p0: UUID?): Int {
         TODO("Not yet implemented")
     }
+
     override fun getVersion(): String = CloudPlatform.pluginVersion
     override fun getSupportedVersions(): SortedSet<Int> = ProtocolRegistry.getSupportedVersions()
 }
@@ -87,12 +91,14 @@ object CloudPlatform : ViaPlatform<Unit> {
     override fun sendMessage(p0: UUID, p1: String) {
         // todo
     }
+
     override fun kickPlayer(p0: UUID, p1: String): Boolean = false // todo
     override fun getApi(): ViaAPI<Unit> = CloudAPI
     override fun getDataFolder(): File = File("viaversion")
     override fun getConf(): ViaVersionConfig = CloudConfig
     override fun onReload() {
     }
+
     override fun getDump(): JsonObject = JsonObject()
     override fun runSync(runnable: Runnable): TaskId = CloudTask(eventLoop.submit(runnable))
     override fun runSync(p0: Runnable, p1: Long): TaskId = CloudTask(eventLoop.schedule(p0, p1 * 50L, TimeUnit.MILLISECONDS))
@@ -105,10 +111,12 @@ object CloudPlatform : ViaPlatform<Unit> {
     override fun cancelTask(p0: TaskId?) {
         (p0 as CloudTask).obj.cancel(false)
     }
+
     override fun isPluginEnabled(): Boolean = true
     override fun getConfigurationProvider(): ConfigurationProvider {
         TODO("Not yet implemented")
     }
+
     override fun getPlatformName(): String = "VIAaaS"
     override fun getPluginVersion(): String = VersionInfo.VERSION
     override fun isOldClientsAllowed(): Boolean = true
@@ -134,9 +142,9 @@ object CloudConfig : AbstractViaConfig(File("config/viaversion.yml")) {
 
     // Based on Sponge ViaVersion
     private val UNSUPPORTED = listOf("anti-xray-patch", "bungee-ping-interval",
-                "bungee-ping-save", "bungee-servers", "quick-move-action-fix", "nms-player-ticking",
-                "item-cache", "velocity-ping-interval", "velocity-ping-save", "velocity-servers",
-                "blockconnection-method", "change-1_9-hitbox", "change-1_14-hitbox")
+            "bungee-ping-save", "bungee-servers", "quick-move-action-fix", "nms-player-ticking",
+            "item-cache", "velocity-ping-interval", "velocity-ping-save", "velocity-servers",
+            "blockconnection-method", "change-1_9-hitbox", "change-1_14-hitbox")
 
     init {
         // Load config
@@ -155,7 +163,7 @@ object CloudConsoleSender : ViaCommandSender {
     override fun hasPermission(p0: String): Boolean = true
 }
 
-object CloudVersionProvider: VersionProvider() {
+object CloudVersionProvider : VersionProvider() {
     override fun getServerProtocol(connection: UserConnection): Int {
         val data = connection.get(CloudData::class.java)
         val ver = data?.backendVer
@@ -164,5 +172,9 @@ object CloudVersionProvider: VersionProvider() {
     }
 }
 
-data class CloudData(val userConnection: UserConnection, val backendAddr: String, val backendVer: Int)
-    : StoredObject(userConnection)
+data class CloudData(val userConnection: UserConnection,
+                     var backendVer: Int,
+                     var backendChannel: SocketChannel? = null,
+                     var frontOnline: Boolean,
+                     var pendingStatus: Boolean = false
+) : StoredObject(userConnection)

src/main/kotlin/com/github/creeper123123321/viaaas/CloudProtocol.kt

@@ -0,0 +1,154 @@
+package com.github.creeper123123321.viaaas
+
+import io.netty.bootstrap.Bootstrap
+import io.netty.buffer.ByteBufAllocator
+import io.netty.channel.Channel
+import io.netty.channel.socket.SocketChannel
+import io.netty.channel.socket.nio.NioSocketChannel
+import us.myles.ViaVersion.api.PacketWrapper
+import us.myles.ViaVersion.api.Via
+import us.myles.ViaVersion.api.protocol.ProtocolVersion
+import us.myles.ViaVersion.api.protocol.SimpleProtocol
+import us.myles.ViaVersion.api.remapper.PacketRemapper
+import us.myles.ViaVersion.api.type.Type
+import us.myles.ViaVersion.packets.State
+import java.net.InetAddress
+
+class CloudHandlerProtocol : SimpleProtocol() {
+    override fun registerPackets() {
+        this.registerIncoming(State.HANDSHAKE, 0, 0, object : PacketRemapper() {
+            override fun registerMap() {
+                handler { wrapper: PacketWrapper ->
+                    val protVer = wrapper.passthrough(Type.VAR_INT)
+                    val addr = wrapper.passthrough(Type.STRING) // Server Address
+                    val svPort = wrapper.passthrough(Type.UNSIGNED_SHORT)
+                    val nextState = wrapper.passthrough(Type.VAR_INT)
+
+                    val addrParts = addr.split(0.toChar())[0].split(".")
+                    var foundDomain = false
+                    var foundOptions = false
+                    var port = 25565
+                    var online = true // todo implement this between proxy and player
+                    var backProtocol = 47 // todo auto protocol
+                    var backAddr = ""
+                    addrParts.reversed().forEach {
+                        if (foundDomain) {
+                            if (!foundOptions) {
+                                if (it.startsWith("_")) {
+                                    val arg = it.substring(2)
+                                    when {
+                                        it.startsWith("_p", ignoreCase = true) -> port = arg.toInt()
+                                        it.startsWith("_o", ignoreCase = true) -> online = arg.toBoolean()
+                                        it.startsWith("_v", ignoreCase = true) -> {
+                                            try {
+                                                backProtocol = Integer.parseInt(arg)
+                                            } catch (e: NumberFormatException) {
+                                                val closest = ProtocolVersion.getClosest(arg.replace("_", "."))
+                                                if (closest != null) {
+                                                    backProtocol = closest.id
+                                                }
+                                            }
+                                        }
+                                    }
+                                } else {
+                                    foundOptions = true
+                                }
+                            }
+                            if (foundOptions) {
+                                backAddr = "$it.$backAddr"
+                            }
+                        } else if (it.equals("viaaas", ignoreCase = true)) {
+                            foundDomain = true
+                        }
+                    }
+
+                    println("connecting ${wrapper.user().channel!!.remoteAddress()} ($protVer) to $backAddr:$port ($backProtocol)")
+
+                    wrapper.user().channel!!.setAutoRead(false)
+                    wrapper.user().put(CloudData(
+                            backendVer = backProtocol,
+                            userConnection = wrapper.user(),
+                            frontOnline = online
+                    ))
+
+                    Via.getPlatform().runAsync {
+                        val frontForwarder = wrapper.user().channel!!.pipeline().get(CloudSideForwarder::class.java)
+                        try {
+                            val backInetAddr = InetAddress.getByName(backAddr)
+                            if (backInetAddr.isAnyLocalAddress) throw SecurityException("Local addresses aren't allowed")
+                            val bootstrap = Bootstrap().handler(BackendInit(wrapper.user()))
+                                    .channel(NioSocketChannel::class.java)
+                                    .group(wrapper.user().channel!!.eventLoop())
+                                    .connect(backInetAddr, port)
+                            println(backInetAddr)
+
+                            bootstrap.addListener {
+                                if (it.isSuccess) {
+                                    val chann = bootstrap.channel() as SocketChannel
+                                    chann.pipeline().get(CloudSideForwarder::class.java).other = wrapper.user().channel
+                                    frontForwarder.other = chann
+                                    val backHandshake = ByteBufAllocator.DEFAULT.buffer()
+                                    try {
+                                        backHandshake.writeByte(0) // Packet 0 handshake
+                                        Type.VAR_INT.writePrimitive(backHandshake, protVer) // client ver
+                                        val nullPos = addr.indexOf(0.toChar())
+                                        Type.STRING.write(backHandshake, backAddr
+                                                + (if (nullPos != -1) addr.substring(nullPos) else "")) // Server Address
+                                        backHandshake.writeShort(port)
+                                        Type.VAR_INT.writePrimitive(backHandshake, nextState)
+                                        chann.writeAndFlush(backHandshake.retain())
+                                    } finally {
+                                        backHandshake.release()
+                                    }
+                                } else {
+                                    wrapper.user().channel!!.eventLoop().submit {
+                                        frontForwarder.disconnect("Couldn't connect: " + it.cause().toString())
+                                    }
+                                }
+
+                                wrapper.user().channel!!.setAutoRead(true)
+                            }
+                        } catch (e: Exception) {
+                            wrapper.user().channel!!.eventLoop().submit {
+                                frontForwarder.disconnect("Couldn't connect: $e")
+                            }
+                        }
+                    }
+                }
+            }
+        })
+
+        this.registerOutgoing(State.LOGIN, 3, 3, object : PacketRemapper() {
+            // set compression
+            override fun registerMap() {
+                handler {
+                    val pipe = it.user().channel!!.pipeline()
+                    val threshold = it.passthrough(Type.VAR_INT)
+                    pipe.get(CloudCompressor::class.java).threshold = threshold
+                    pipe.get(CloudDecompressor::class.java).threshold = threshold
+
+                    val backPipe = it.user().channel!!
+                            .pipeline().get(CloudSideForwarder::class.java).other?.pipeline()
+                    backPipe?.get(CloudCompressor::class.java)?.threshold = threshold
+                    backPipe?.get(CloudDecompressor::class.java)?.threshold = threshold
+                }
+            }
+        })
+
+        this.registerOutgoing(State.LOGIN, 1, 1, object : PacketRemapper() {
+            // encryption request
+            override fun registerMap() {
+                handler {
+                    val frontForwarder = it.user().channel!!.pipeline().get(CloudSideForwarder::class.java)
+                    it.cancel()
+                    frontForwarder.disconnect("Online mode in backend currently isn't compatible")
+                }
+            }
+        })
+    }
+}
+
+fun Channel.setAutoRead(b: Boolean) {
+    this.config().isAutoRead = b
+    if (b) this.read()
+}

src/main/kotlin/com/github/creeper123123321/viaaas/ViaaaS.kt

@@ -6,10 +6,8 @@ import io.netty.channel.socket.nio.NioServerSocketChannel
 import us.myles.ViaVersion.ViaManager
 import us.myles.ViaVersion.api.Via
 import us.myles.ViaVersion.api.data.MappingDataLoader
-import us.myles.ViaVersion.api.data.UserConnection
 
-
-fun main(args: Array<String>) {
+fun main() {
     Via.init(ViaManager.builder()
             .injector(CloudInjector)
             .loader(CloudLoader)
@@ -19,12 +17,15 @@ fun main(args: Array<String>) {
     MappingDataLoader.enableMappingsCache()
 
     Via.getManager().init()
+
     val boss = NioEventLoopGroup()
     val worker = NioEventLoopGroup()
     val future = ServerBootstrap().group(boss, worker)
             .channel(NioServerSocketChannel::class.java)
             .childHandler(ChannelInit)
             .bind(25565)
+            .addListener { println(it) }
+
 
     loop@ while (true) {
         try {