first commit

.github/workflows/rust.yml

@@ -0,0 +1,80 @@
+name: Cargo Build, Test, and Linting
+
+on:
+  push:
+  pull_request:
+
+env:
+  CARGO_TERM_COLOR: always
+  RUSTFLAGS: "-Dwarnings"
+
+jobs:
+  format:
+    name: Check formatting
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        toolchain:
+          - stable
+    steps:
+      - uses: actions/checkout@v4
+      - run: rustup update ${{ matrix.toolchain }} && rustup default ${{ matrix.toolchain }}
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo fmt --check
+  clippy:
+    name: Run lints
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        toolchain:
+          - stable
+    steps:
+      - uses: actions/checkout@v4
+      - run: rustup update ${{ matrix.toolchain }} && rustup default ${{ matrix.toolchain }}
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo clippy --all-targets --all-features
+  build_and_test:
+    name: Build project and test
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ubuntu-latest, windows-latest, macos-latest]
+        toolchain:
+          - stable
+    steps:
+      - uses: actions/checkout@v4
+      - run: rustup update ${{ matrix.toolchain }} && rustup default ${{ matrix.toolchain }}
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo test --verbose
+  build_release:
+    name: Build project in release
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ubuntu-latest, windows-latest, macos-latest]
+        toolchain:
+          - stable
+    steps:
+      - uses: actions/checkout@v4
+
+      - run: rustup update ${{ matrix.toolchain }} && rustup default ${{ matrix.toolchain }}
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo build --verbose --release
+      - name: Export executable
+        uses: actions/upload-artifact@v4
+        with:
+          name: pumpkin-${{ matrix.os }}
+          compression-level: 9
+          path: target/${{ matrix.target }}/release/pumpkin*
+  clippy_release:
+    name: Run lints in release mode
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        toolchain:
+          - stable
+    steps:
+      - uses: actions/checkout@v4
+      - run: rustup update ${{ matrix.toolchain }} && rustup default ${{ matrix.toolchain }}
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo clippy --release --all-targets --all-features

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "botmark"
+version = "0.1.0"
+edition = "2024"
+
+[profile.release]
+lto = true
+codegen-units = 1
+
+[dependencies]
+# log
+log = "0.4"
+simple_logger = "5.0.0"
+# CLI
+clap = { version = "4.5.30", features = ["derive"] }
+# Networking
+tokio = { version = "1.43.0", features = ["net", "io-util"] }
+pumpkin-protocol = { git = "https://github.com/Pumpkin-MC/Pumpkin", package = "pumpkin-protocol", default-features = false, features = ["packets"]}
+
+crossbeam = "0.8.4"
+uuid = "1.14.0"

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 Pumpkin MC
+
+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
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,16 @@
+<div align="center">
+
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
+![Current version)](https://img.shields.io/badge/current_version-1.21.4-blue)
+
+A tool to stress test Minecraft servers by simulating numerous bot connections, helping identify performance bottlenecks and stability issues.
+![Screenshot_20250223_154536](https://github.com/user-attachments/assets/b1379060-e375-44dc-95a3-91f1f2fb52dd)
+</div>
+
+### How to Run
+1. **Disable Online Mode:** Ensure that `online-mode=false` in your server's `server.properties` file.  This is essential for Botmark to connect.
+
+2. **Run Botmark:** Use the following command in your terminal:
+`./botmark --ip <ip:port> --count <bot-count>`
+**Example**
+`./botmark --ip 0.0.0.0:25565 --count 50`

src/client.rs

@@ -0,0 +1,281 @@
+use crossbeam::atomic::AtomicCell;
+use pumpkin_protocol::bytebuf::ReadingError;
+use pumpkin_protocol::bytebuf::packet::Packet;
+use pumpkin_protocol::client::config::{CConfigDisconnect, CFinishConfig};
+use pumpkin_protocol::client::login::{
+    CEncryptionRequest, CLoginDisconnect, CLoginSuccess, CSetCompression,
+};
+use pumpkin_protocol::client::play::{CKeepAlive, CPlayDisconnect, CPlayerPosition};
+use pumpkin_protocol::codec::var_int::VarInt;
+use pumpkin_protocol::server::config::{SAcknowledgeFinishConfig, SKnownPacks};
+use pumpkin_protocol::server::handshake::SHandShake;
+use pumpkin_protocol::server::login::{SLoginAcknowledged, SLoginStart};
+use pumpkin_protocol::server::play::{SConfirmTeleport, SKeepAlive};
+use pumpkin_protocol::{CURRENT_MC_PROTOCOL, RawPacket, ServerPacket};
+use pumpkin_protocol::{
+    ClientPacket, CompressionLevel, CompressionThreshold, ConnectionState,
+    packet_decoder::PacketDecoder, packet_encoder::PacketEncoder,
+};
+use std::collections::VecDeque;
+use std::net::SocketAddr;
+use std::sync::{Arc, atomic::AtomicBool};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::{
+    net::{
+        TcpStream,
+        tcp::{OwnedReadHalf, OwnedWriteHalf},
+    },
+    sync::Mutex,
+};
+use uuid::Uuid;
+
+/// Everything which makes a Connection with our Server is a `Client`.
+pub struct Client {
+    /// The current connection state of the client (e.g., Handshaking, Status, Play).
+    pub connection_state: AtomicCell<ConnectionState>,
+    /// Indicates if the client connection is closed.
+    pub closed: AtomicBool,
+    /// The packet encoder for outgoing packets.
+    pub enc: Arc<Mutex<PacketEncoder>>,
+    /// The packet decoder for incoming packets.
+    pub dec: Arc<Mutex<PacketDecoder>>,
+    pub connection_reader: Mutex<OwnedReadHalf>,
+    pub connection_writer: Mutex<OwnedWriteHalf>,
+    pub client_packets_queue: Arc<Mutex<VecDeque<RawPacket>>>,
+}
+
+impl Client {
+    pub fn new(stream: TcpStream) -> Self {
+        let (connection_reader, connection_writer) = stream.into_split();
+        Self {
+            connection_state: AtomicCell::new(ConnectionState::HandShake),
+            enc: Arc::new(Mutex::new(PacketEncoder::default())),
+            dec: Arc::new(Mutex::new(PacketDecoder::default())),
+            closed: AtomicBool::new(false),
+            connection_reader: Mutex::new(connection_reader),
+            connection_writer: Mutex::new(connection_writer),
+            client_packets_queue: Arc::new(Mutex::new(VecDeque::new())),
+        }
+    }
+
+    /// Enables or disables packet compression for the connection.
+    ///
+    /// This function takes an optional `CompressionInfo` struct as input. If the `CompressionInfo` is provided,
+    /// packet compression is enabled with the specified threshold. Otherwise, compression is disabled.
+    ///
+    /// # Arguments
+    ///
+    /// * `compression`: An optional `CompressionInfo` struct containing the compression threshold and compression level.
+    pub async fn set_compression(
+        &self,
+        compression: Option<(CompressionThreshold, CompressionLevel)>,
+    ) {
+        self.dec.lock().await.set_compression(compression.is_some());
+        self.enc
+            .lock()
+            .await
+            .set_compression(compression.map(|s| (s.0, s.1)))
+            .unwrap_or_else(|_| log::warn!("invalid compression level"));
+    }
+
+    pub async fn process_packets(&self) {
+        let mut packet_queue = self.client_packets_queue.lock().await;
+        while let Some(mut packet) = packet_queue.pop_front() {
+            if self.closed.load(std::sync::atomic::Ordering::Relaxed) {
+                log::debug!("Canceling client packet processing (pre)");
+                return;
+            }
+            if let Err(error) = self.handle_packet(&mut packet).await {
+                log::error!(
+                    "Failed to read incoming packet with id {}: {}",
+                    i32::from(packet.id),
+                    error
+                );
+                self.close().await;
+            };
+        }
+    }
+    pub async fn add_packet(&self, packet: RawPacket) {
+        let mut client_packets_queue = self.client_packets_queue.lock().await;
+        client_packets_queue.push_back(packet);
+    }
+
+    pub async fn poll(&self) -> bool {
+        loop {
+            if self.closed.load(std::sync::atomic::Ordering::Relaxed) {
+                // If we manually close (like a kick) we dont want to keep reading bytes
+                return false;
+            }
+
+            let mut dec = self.dec.lock().await;
+
+            match dec.decode() {
+                Ok(Some(packet)) => {
+                    self.add_packet(packet).await;
+                    return true;
+                }
+                Ok(None) => (), //log::debug!("Waiting for more data to complete packet..."),
+                Err(err) => {
+                    log::warn!("Failed to decode packet for: {}", err.to_string());
+                    self.close().await;
+                    return false; // return to avoid reserving additional bytes
+                }
+            }
+
+            dec.reserve(4096);
+            let mut buf = dec.take_capacity();
+
+            let bytes_read = self.connection_reader.lock().await.read_buf(&mut buf).await;
+            match bytes_read {
+                Ok(cnt) => {
+                    //log::debug!("Read {} bytes", cnt);
+                    if cnt == 0 {
+                        self.close().await;
+                        return false;
+                    }
+                }
+                Err(error) => {
+                    log::error!("Error while reading incoming packet {}", error);
+                    self.close().await;
+                    return false;
+                }
+            };
+
+            // This should always be an O(1) unsplit because we reserved space earlier and
+            // the call to `read_buf` shouldn't have grown the allocation.
+            dec.queue_bytes(buf);
+        }
+    }
+
+    /// Sends a clientbound packet to the connected client.
+    ///
+    /// # Arguments
+    ///
+    /// * `packet`: A reference to a packet object implementing the `ClientPacket` trait.
+    pub async fn send_packet<P: ClientPacket>(&self, packet: &P) {
+        //log::debug!("Sending packet with id {} to {}", P::PACKET_ID, self.id);
+        // assert!(!self.closed);
+        if self.closed.load(std::sync::atomic::Ordering::Relaxed) {
+            return;
+        }
+
+        let mut enc = self.enc.lock().await;
+        if let Err(_error) = enc.append_packet(packet) {
+            return;
+        }
+
+        let mut writer = self.connection_writer.lock().await;
+        if let Err(error) = writer.write_all(&enc.take()).await {
+            log::debug!("Unable to write to connection: {}", error.to_string());
+        }
+    }
+
+    pub async fn join_server(&self, address: SocketAddr, name: String) {
+        self.send_packet(&SHandShake {
+            protocol_version: VarInt(CURRENT_MC_PROTOCOL.get() as i32),
+            server_address: address.ip().to_string(),
+            server_port: address.port(),
+            next_state: pumpkin_protocol::ConnectionState::Login,
+        })
+        .await;
+        self.connection_state.store(ConnectionState::Login);
+        self.send_packet(&SLoginStart {
+            name,
+            uuid: Uuid::new_v4(),
+        })
+        .await;
+    }
+
+    pub async fn handle_packet(&self, packet: &mut RawPacket) -> Result<(), ReadingError> {
+        match self.connection_state.load() {
+            ConnectionState::HandShake => unreachable!(),
+            ConnectionState::Status => todo!(),
+            ConnectionState::Login => self.handle_login_packet(packet).await?,
+            ConnectionState::Transfer => log::debug!("Got packet in transfer state"),
+            ConnectionState::Config => self.handle_config_packet(packet).await?,
+            ConnectionState::Play => self.handle_play_packet(packet).await?,
+        };
+        Ok(())
+    }
+
+    async fn handle_login_packet(&self, packet: &mut RawPacket) -> Result<(), ReadingError> {
+        let bytebuf = &mut packet.bytebuf;
+        match packet.id.0 {
+            CEncryptionRequest::PACKET_ID => {
+                log::debug!("Got Encryption Request")
+            }
+            CSetCompression::PACKET_ID => {
+                log::trace!("Set Compression");
+                let packet = CSetCompression::read(bytebuf)?;
+                self.set_compression(Some((
+                    CompressionThreshold(packet.threshold.0 as u32),
+                    CompressionLevel(6),
+                )))
+                .await
+            }
+            CLoginDisconnect::PACKET_ID => {
+                log::error!("Kicking in Login State");
+                self.close().await;
+            }
+            CLoginSuccess::PACKET_ID => {
+                log::trace!("Login -> Config");
+                self.send_packet(&SLoginAcknowledged).await;
+                self.connection_state.store(ConnectionState::Config);
+                log::trace!("Sending Known packs");
+                self.send_packet(&SKnownPacks {
+                    known_pack_count: VarInt(0),
+                })
+                .await;
+            }
+            _ => {}
+        }
+        Ok(())
+    }
+
+    async fn handle_config_packet(&self, packet: &mut RawPacket) -> Result<(), ReadingError> {
+        match packet.id.0 {
+            CConfigDisconnect::PACKET_ID => {
+                log::error!("Kicking in Config State");
+                self.close().await;
+            }
+            CFinishConfig::PACKET_ID => {
+                log::trace!("Config -> Play");
+                self.send_packet(&SAcknowledgeFinishConfig).await;
+                self.connection_state.store(ConnectionState::Play);
+            }
+            _ => {}
+        }
+        Ok(())
+    }
+
+    async fn handle_play_packet(&self, packet: &mut RawPacket) -> Result<(), ReadingError> {
+        let bytebuf = &mut packet.bytebuf;
+        match packet.id.0 {
+            CKeepAlive::PACKET_ID => {
+                let packet = CKeepAlive::read(bytebuf)?;
+                self.send_packet(&SKeepAlive {
+                    keep_alive_id: packet.keep_alive_id,
+                })
+                .await;
+            }
+            CPlayerPosition::PACKET_ID => {
+                let packet = CPlayerPosition::read(bytebuf)?;
+                self.send_packet(&SConfirmTeleport {
+                    teleport_id: packet.teleport_id,
+                })
+                .await;
+            }
+            CPlayDisconnect::PACKET_ID => {
+                log::error!("Kicking in Play State");
+                self.close().await;
+            }
+            _ => {}
+        }
+        Ok(())
+    }
+
+    pub async fn close(&self) {
+        self.closed
+            .store(true, std::sync::atomic::Ordering::Relaxed);
+    }
+}

src/main.rs

@@ -0,0 +1,51 @@
+use std::net::SocketAddr;
+
+use clap::Parser;
+use client::Client;
+use tokio::net::TcpStream;
+
+mod client;
+
+#[derive(Parser, Debug)]
+#[command(version, about, long_about = None)]
+struct Args {
+    #[arg(short, long)]
+    ip: String,
+    #[arg(short, long, default_value_t = 1)]
+    count: u32,
+}
+
+#[tokio::main]
+async fn main() {
+    simple_logger::init_with_level(log::Level::Info).unwrap();
+    let args = Args::parse();
+    let address = args.ip.parse::<SocketAddr>().unwrap();
+    log::info!("{} Bots will Join {}", args.count, address);
+    let mut join_handles = Vec::with_capacity(args.count as usize);
+
+    for i in 0..args.count {
+        let stream = TcpStream::connect(address)
+            .await
+            .expect("Failed to connect to Ip");
+        let client = Client::new(stream);
+        client.join_server(address, format!("BOT_{}", i)).await;
+        let join_handle = tokio::spawn(async move {
+            loop {
+                if !client.poll().await {
+                    break;
+                }
+                client.process_packets().await;
+            }
+        });
+        join_handles.push(join_handle);
+        log::info!("{}/{} Bots Joined", i + 1, args.count);
+    }
+    // Graceful shutdown on Ctrl+C
+    tokio::signal::ctrl_c().await.unwrap();
+    log::info!("Shutting down...");
+
+    // Wait for all bot tasks to exit
+    for handle in join_handles {
+        handle.abort();
+    }
+}