Ext-encoding support

Author: dries-c

.github/workflows/phpstan.yml

@@ -5,7 +5,7 @@ on: push
 jobs:
   phpstan:
     name: PHPStan Analysis
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
     if: "!contains(github.event.head_commit.message, '[ci skip]')"
 
     steps:
@@ -14,14 +14,14 @@ jobs:
       - name: Download PHP Release
         uses: dsaltares/fetch-gh-release-asset@1.1.2
         with:
-          file: PHP-8.3-Linux-x86_64-PM5.tar.gz
+          file: PHP-8.4-Linux-x86_64-PM5.tar.gz
           repo: NetherGamesMC/php-build-scripts
-          version: "tags/pm5-php-8.3-latest"
+          version: "tags/pm5-php-8.4-latest"
           token: ${{ secrets.GITHUB_TOKEN }}
       - name: Unpack PHP Release
-        run: tar -xzvf PHP-8.3-Linux-x86_64-PM5.tar.gz
-      - name: Install libFFI
-        run: sudo apt install libffi7
+        run: tar -xzvf PHP-8.4-Linux-x86_64-PM5.tar.gz
+      - name: Install libffi7
+        run: sudo apt update && sudo apt install -y --no-install-recommends libffi7
       - name: Download Composer
         run: curl -o composer.phar "https://getcomposer.org/composer-stable.phar"
       - name: Add Composer GitHub access token

ProxyNetworkInterface.php

@@ -15,9 +15,13 @@
 use libproxy\protocol\LoginPacket;
 use libproxy\protocol\ProxyPacket;
 use libproxy\protocol\ProxyPacketPool;
-use libproxy\protocol\ProxyPacketSerializer;
+use pmmp\encoding\ByteBufferReader;
+use pmmp\encoding\ByteBufferWriter;
+use pmmp\encoding\DataDecodeException;
+use pmmp\encoding\LE;
 use pmmp\thread\Thread as NativeThread;
 use pmmp\thread\ThreadSafeArray;
+use pocketmine\network\FilterNoisyPacketException;
 use pocketmine\network\mcpe\compression\ZlibCompressor;
 use pocketmine\network\mcpe\convert\TypeConverter;
 use pocketmine\network\mcpe\EntityEventBroadcaster;
@@ -33,8 +37,6 @@
 use pocketmine\Server;
 use pocketmine\snooze\SleeperHandlerEntry;
 use pocketmine\thread\ThreadCrashException;
-use pocketmine\utils\Binary;
-use pocketmine\utils\BinaryDataException;
 use Socket;
 use ThreadedArray;
 use WeakMap;
@@ -97,11 +99,9 @@ public function __construct(PluginBase $plugin, int $port, ?string $composerPath
             }
         }
 
-        /** @phpstan-ignore-next-line */
         self::$latencyMap = new WeakMap();
 
-        /** @var Socket $threadNotifier */
-        /** @var Socket $threadNotification */
+        /** @var list{Socket, Socket} $ipc */
         [$threadNotifier, $threadNotification] = $ipc;
         $this->threadNotifier = $threadNotifier;
 
@@ -163,27 +163,29 @@ public function start(): void
 
     /**
      * @throws PacketHandlingException
+     * @throws DataDecodeException
      */
     private function onPacketReceive(string $buffer): void
     {
-        $stream = new ProxyPacketSerializer($buffer);
-        $socketId = $stream->getLInt();
+        $stream = new ByteBufferReader($buffer);
+        $socketId = LE::readUnsignedInt($stream);
 
-        if (($pk = ProxyPacketPool::getInstance()->getPacket($buffer, $stream->getOffset())) === null) {
-            $offset = 0;
-            throw new PacketHandlingException('Proxy packet with id (' . Binary::readUnsignedVarInt($buffer, $offset) . ') does not exist');
+        $offset = $stream->getOffset();
+        if (($pk = ProxyPacketPool::getInstance()->getPacket($stream)) === null) {
+            throw new PacketHandlingException('Unknown ProxyPacket received from Proxy Thread');
         }
+        $stream->setOffset($offset);
 
         try {
             $pk->decode($stream);
-        } catch (BinaryDataException $e) {
+        } catch (DataDecodeException $e) {
             $this->server->getLogger()->debug('Closed socket with id(' . $socketId . ') because packet was invalid.');
             $this->close($socketId, 'Invalid Packet');
             return;
         }
 
-        if (!$stream->feof()) {
-            $remains = substr($stream->getBuffer(), $stream->getOffset());
+        if ($stream->getUnreadLength() > 0) {
+            $remains = substr($stream->getData(), $stream->getOffset());
             $this->server->getLogger()->debug('Still ' . strlen($remains) . ' bytes unread in ' . $pk->pid() . ': ' . bin2hex($remains));
         }
 
@@ -211,6 +213,10 @@ private function onPacketReceive(string $buffer): void
                         break; // might be data arriving from the client after the server has closed the connection
                     }
 
+                    if ((fn() => $this->checkRepeatedPacketFilter($pk->payload))->call($session)) {
+                        break;
+                    }
+
                     $packet = PacketPool::getInstance()->getPacket($pk->payload);
                     if ($packet === null) {
                         $session->getLogger()->debug("Unknown packet: " . base64_encode($pk->payload));
@@ -221,6 +227,8 @@ private function onPacketReceive(string $buffer): void
                     } catch (PacketHandlingException $e) {
                         $session->getLogger()->debug($packet->getName() . ": " . base64_encode($pk->payload));
                         throw PacketHandlingException::wrap($e, "Error processing " . $packet->getName());
+                    } catch (FilterNoisyPacketException) {
+                        (fn() => $this->noisyPacketBuffer = $pk->payload)->call($session);
                     }
                     $this->receiveBytes += strlen($pk->payload);
                     break;
@@ -233,7 +241,7 @@ private function onPacketReceive(string $buffer): void
                     $session->handleAckReceipt($pk->receiptId);
                     break;
             }
-        } catch (PacketHandlingException|BinaryDataException $exception) {
+        } catch (PacketHandlingException|DataDecodeException $exception) {
             $this->close($socketId, 'Error handling a Packet (Server)');
 
             $this->server->getLogger()->logException($exception);
@@ -285,13 +293,13 @@ public function getSession(int $socketId): ?NetworkSession
 
     public function putPacket(int $socketId, ProxyPacket $pk): void
     {
-        $serializer = new ProxyPacketSerializer();
-        $serializer->putLInt($socketId);
+        $serializer = new ByteBufferWriter();
+        LE::writeUnsignedInt($serializer, $socketId);
 
         $pk->encode($serializer);
 
-        $this->mainToThreadWriter->write($serializer->getBuffer());
-        $this->sendBytes += strlen($serializer->getBuffer());
+        $this->mainToThreadWriter->write($serializer->getData());
+        $this->sendBytes += strlen($serializer->getData());
 
         try {
             socket_write($this->threadNotifier, "\x00"); // wakes up the socket_select function

ProxyServer.php

@@ -12,13 +12,17 @@
 use libproxy\protocol\LoginPacket;
 use libproxy\protocol\ProxyPacket;
 use libproxy\protocol\ProxyPacketPool;
-use libproxy\protocol\ProxyPacketSerializer;
 use NetherGames\Quiche\io\QueueWriter;
 use NetherGames\Quiche\QuicheConnection;
 use NetherGames\Quiche\socket\QuicheServerSocket;
 use NetherGames\Quiche\SocketAddress;
 use NetherGames\Quiche\stream\BiDirectionalQuicheStream;
 use NetherGames\Quiche\stream\QuicheStream;
+use pmmp\encoding\BE;
+use pmmp\encoding\ByteBufferReader;
+use pmmp\encoding\ByteBufferWriter;
+use pmmp\encoding\DataDecodeException;
+use pmmp\encoding\LE;
 use pmmp\thread\ThreadSafeArray;
 use pocketmine\network\mcpe\compression\DecompressionException;
 use pocketmine\network\mcpe\compression\ZlibCompressor;
@@ -31,17 +35,13 @@
 use pocketmine\network\mcpe\protocol\ProtocolInfo;
 use pocketmine\network\mcpe\protocol\RequestNetworkSettingsPacket;
 use pocketmine\network\mcpe\protocol\serializer\PacketBatch;
-use pocketmine\network\mcpe\protocol\serializer\PacketSerializer;
 use pocketmine\network\mcpe\protocol\types\CompressionAlgorithm;
 use pocketmine\network\mcpe\raklib\PthreadsChannelReader;
 use pocketmine\network\mcpe\raklib\SnoozeAwarePthreadsChannelWriter;
 use pocketmine\network\PacketHandlingException;
 use pocketmine\snooze\SleeperHandler;
 use pocketmine\snooze\SleeperHandlerEntry;
 use pocketmine\thread\log\AttachableThreadSafeLogger;
-use pocketmine\utils\Binary;
-use pocketmine\utils\BinaryDataException;
-use pocketmine\utils\BinaryStream;
 use Socket;
 use function array_keys;
 use function base64_encode;
@@ -56,6 +56,8 @@
 
 class ProxyServer
 {
+    private const INCOMING_PACKET_BATCH_HARD_LIMIT = 300;
+
     /** @var PthreadsChannelReader */
     private PthreadsChannelReader $mainToThreadReader;
     /** @var SnoozeAwarePthreadsChannelWriter */
@@ -221,12 +223,12 @@ private function shutdownStream(int $streamIdentifier, string $reason, bool $fro
 
     private function sendToMainBuffer(int $streamIdentifier, ProxyPacket $pk): void
     {
-        $serializer = new ProxyPacketSerializer();
-        $serializer->putLInt($streamIdentifier);
+        $serializer = new ByteBufferWriter();
+        LE::writeUnsignedInt($serializer, $streamIdentifier);
 
         $pk->encode($serializer);
 
-        $this->threadToMainWriter->write($serializer->getBuffer());
+        $this->threadToMainWriter->write($serializer->getData());
     }
 
     public function tickProcessor(): void
@@ -237,16 +239,18 @@ public function tickProcessor(): void
     private function pushSockets(): void
     {
         while (($payload = $this->mainToThreadReader->read()) !== null) {
-            $stream = new ProxyPacketSerializer($payload);
-            $streamIdentifier = $stream->getLInt();
+            $stream = new ByteBufferReader($payload);
+            $streamIdentifier = LE::readUnsignedInt($stream);
 
-            if (($pk = ProxyPacketPool::getInstance()->getPacket($payload, $stream->getOffset())) === null) {
+            $offset = $stream->getOffset();
+            if (($pk = ProxyPacketPool::getInstance()->getPacket($stream)) === null) {
                 throw new PacketHandlingException('Packet does not exist');
             }
+            $stream->setOffset($offset);
 
             try {
                 $pk->decode($stream);
-            } catch (BinaryDataException $e) {
+            } catch (DataDecodeException $e) {
                 $this->logger->debug('Closed stream with id(' . $streamIdentifier . ') because server sent invalid packet');
                 $this->shutdownStream($streamIdentifier, 'invalid packet', false);
                 return;
@@ -278,7 +282,7 @@ private function sendPayloadWithReceipt(int $streamIdentifier, string $payload,
             return;
         }
 
-        $writer->writeWithPromise(Binary::writeInt(strlen($payload)) . $payload)->onResult(function() use ($streamIdentifier, $receiptId): void{
+        $writer->writeWithPromise(BE::packSignedInt(strlen($payload)) . $payload)->onResult(function () use ($streamIdentifier, $receiptId): void {
             $pk = new AckPacket();
             $pk->receiptId = $receiptId;
 
@@ -296,7 +300,7 @@ private function sendPayload(int $streamIdentifier, string $payload): void
             return;
         }
 
-        $writer->write(Binary::writeInt(strlen($payload)) . $payload);
+        $writer->write(BE::packSignedInt(strlen($payload)) . $payload);
     }
 
     /**
@@ -323,26 +327,26 @@ private function getProtocolId(int $streamIdentifier): int
      */
     private function sendDataPacket(int $streamIdentifier, BedrockPacket $packet): void
     {
-        $packetSerializer = PacketSerializer::encoder($protocolId = $this->getProtocolId($streamIdentifier));
-        $packet->encode($packetSerializer);
+        $packetSerializer = new ByteBufferWriter();
+        $packet->encode($packetSerializer, $protocolId = $this->getProtocolId($streamIdentifier));
 
-        $stream = new BinaryStream();
-        PacketBatch::encodeRaw($stream, [$packetSerializer->getBuffer()]);
-        $payload = ($protocolId >= ProtocolInfo::PROTOCOL_1_20_60 ? chr(CompressionAlgorithm::ZLIB) : '') . ZlibCompressor::getInstance()->compress($stream->getBuffer());
+        $stream = new ByteBufferWriter();
+        PacketBatch::encodeRaw($stream, [$packetSerializer->getData()]);
+        $payload = ($protocolId >= ProtocolInfo::PROTOCOL_1_20_60 ? chr(CompressionAlgorithm::ZLIB) : '') . ZlibCompressor::getInstance()->compress($stream->getData());
 
         $this->sendPayload($streamIdentifier, $payload);
     }
 
     private function decodePacket(int $streamIdentifier, BedrockPacket $packet, string $buffer): void
     {
-        $stream = PacketSerializer::decoder($this->protocolId[$streamIdentifier] ?? ProtocolInfo::CURRENT_PROTOCOL, $buffer, 0);
+        $stream = new ByteBufferReader($buffer);
         try {
-            $packet->decode($stream);
+            $packet->decode($stream, $this->protocolId[$streamIdentifier] ?? ProtocolInfo::CURRENT_PROTOCOL);
         } catch (PacketDecodeException $e) {
             throw PacketHandlingException::wrap($e);
         }
-        if (!$stream->feof()) {
-            $remains = substr($stream->getBuffer(), $stream->getOffset());
+        if ($stream->getUnreadLength() > 0) {
+            $remains = substr($stream->getData(), $stream->getOffset());
             $this->logger->debug("Still " . strlen($remains) . " bytes unread in " . $packet->getName() . ": " . bin2hex($remains));
         }
     }
@@ -407,14 +411,18 @@ private function onFullDataReceive(int $streamIdentifier, string $payload): void
                 throw PacketHandlingException::wrap($e, "Compressed packet batch decode error");
             }
 
+            $count = 0;
             try {
-                $stream = new BinaryStream($decompressed);
-                $count = 0;
+                $stream = new ByteBufferReader($decompressed);
                 foreach (PacketBatch::decodeRaw($stream) as $buffer) {
-                    $this->getGamePacketLimiter($streamIdentifier)->decrement();
-                    if (++$count > 100) {
-                        throw new PacketHandlingException("Too many packets in batch");
+                    if(++$count >= self::INCOMING_PACKET_BATCH_HARD_LIMIT){
+                        //this should be well more than enough; under normal conditions the game packet rate limiter
+                        //will kick in well before this. This is only here to make sure we can't get huge batches of
+                        //noisy packets to bog down the server, since those aren't counted by the regular limiter.
+                        throw new PacketHandlingException("Reached hard limit of " . self::INCOMING_PACKET_BATCH_HARD_LIMIT . " per batch packet");
                     }
+
+                    $this->getGamePacketLimiter($streamIdentifier)->decrement();
                     $packet = PacketPool::getInstance()->getPacket($buffer);
                     if ($packet === null) {
                         $this->logger->debug("Unknown packet: " . base64_encode($buffer));
@@ -429,7 +437,7 @@ private function onFullDataReceive(int $streamIdentifier, string $payload): void
                         throw PacketHandlingException::wrap($e, "Error processing " . $packet->getName());
                     }
                 }
-            } catch (PacketDecodeException|BinaryDataException $e) {
+            } catch (PacketDecodeException|DataDecodeException $e) {
                 $this->logger->logException($e);
                 throw PacketHandlingException::wrap($e, "Packet batch decode error");
             }
@@ -457,8 +465,8 @@ private function onDataReceive(int $streamIdentifier, string $data): void
                     return; // wait for more data
                 } else {
                     try {
-                        $packetLength = Binary::readInt(substr($buffer, 0, 4));
-                    } catch (BinaryDataException $exception) {
+                        $packetLength = BE::unpackSignedInt(substr($buffer, 0, 4));
+                    } catch (DataDecodeException $exception) {
                         $this->shutdownStream($streamIdentifier, 'invalid packet', false);
                         return;
                     }

composer.json

@@ -5,13 +5,14 @@
     "type": "project",
     "version": "dev-pm5",
     "require": {
-        "php": "^8.0",
-        "ext-sockets": "*",
+        "php": "^8.3",
+        "ext-encoding": "~1.0.0",
         "ext-pmmpthread": "^6.0.1",
+        "ext-sockets": "*",
         "nethergamesmc/quiche": "dev-master"
     },
     "require-dev": {
-        "phpstan/phpstan": "2.1.1",
+        "phpstan/phpstan": "2.1.29",
         "nethergamesmc/pocketmine-mp": "dev-stable"
     },
     "repositories": [

phpstan.neon.dist

@@ -1,6 +1,7 @@
 parameters:
+    ignoreErrors:
+        - identifier: function.alreadyNarrowedType
     level: max
-    checkMissingIterableValueType: false
     paths:
         - .
     excludePaths:

protocol/AckPacket.php

@@ -5,20 +5,23 @@
 
 namespace libproxy\protocol;
 
+use pmmp\encoding\ByteBufferReader;
+use pmmp\encoding\ByteBufferWriter;
+use pmmp\encoding\VarInt;
 
 class AckPacket extends ProxyPacket
 {
-    public const NETWORK_ID = ProxyProtocolInfo::ACK_PACKET;
+    public const int NETWORK_ID = ProxyProtocolInfo::ACK_PACKET;
 
     public int $receiptId;
 
-    public function encodePayload(ProxyPacketSerializer $out): void
+    public function encodePayload(ByteBufferWriter $out): void
     {
-        $out->putUnsignedVarInt($this->receiptId);
+        VarInt::writeUnsignedInt($out, $this->receiptId);
     }
 
-    public function decodePayload(ProxyPacketSerializer $in): void
+    public function decodePayload(ByteBufferReader $in): void
     {
-        $this->receiptId = $in->getUnsignedVarInt();
+        $this->receiptId = VarInt::readUnsignedInt($in);
     }
 }