From f0d343edb1d499ace5edd2ed19368117171317ae Mon Sep 17 00:00:00 2001 From: zhibei <785740487@qq.com> Date: Tue, 14 Jul 2026 01:44:02 +0800 Subject: [PATCH] =?UTF-8?q?fix(porticus):=20=E4=BF=AE=E5=A4=8D=E5=8F=91?= =?UTF-8?q?=E9=80=81=E7=BA=BF=E7=A8=8B=E4=B8=8E=E5=88=86=E5=8C=85=E7=AB=9E?= =?UTF-8?q?=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../minecraft-porticus/build.gradle.kts | 4 + .../module/porticus/PorticusMission.java | 31 +- .../porticus/bukkitside/MissionBukkit.java | 95 ++++- .../porticus/bukkitside/PorticusListener.java | 73 +++- .../porticus/bungeeside/MissionBungee.java | 176 +++++++- .../porticus/bungeeside/PorticusListener.java | 97 +++-- .../module/porticus/common/ByteUtils.java | 16 +- .../module/porticus/common/Message.java | 245 ++++++++++- .../porticus/common/MessageBuilder.java | 41 +- .../module/porticus/common/MessagePacket.java | 12 + .../module/porticus/common/MessageReader.java | 395 +++++++++++++++++- .../taboolib/module/porticus/Porticus.kt | 8 + .../module/porticus/PorticusMissionTest.java | 124 ++++++ .../porticus/common/MessageProtocolTest.java | 331 +++++++++++++++ 14 files changed, 1542 insertions(+), 106 deletions(-) create mode 100644 module/minecraft/minecraft-porticus/src/test/java/taboolib/module/porticus/PorticusMissionTest.java create mode 100644 module/minecraft/minecraft-porticus/src/test/java/taboolib/module/porticus/common/MessageProtocolTest.java diff --git a/module/minecraft/minecraft-porticus/build.gradle.kts b/module/minecraft/minecraft-porticus/build.gradle.kts index 387b0075e..6f683bf05 100644 --- a/module/minecraft/minecraft-porticus/build.gradle.kts +++ b/module/minecraft/minecraft-porticus/build.gradle.kts @@ -5,4 +5,8 @@ dependencies { compileOnly(project(":common-util")) compileOnly(project(":common-env")) compileOnly(project(":common-platform-api")) + testImplementation(project(":common")) + testImplementation(project(":common-util")) + testImplementation(project(":common-platform-api")) + testImplementation("ink.ptms.core:v12004:12004-minimize:mapped") } \ No newline at end of file diff --git a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/PorticusMission.java b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/PorticusMission.java index 73b218555..16d06df76 100644 --- a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/PorticusMission.java +++ b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/PorticusMission.java @@ -6,6 +6,7 @@ import java.util.UUID; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import java.util.function.LongSupplier; /** * Porticus @@ -24,7 +25,9 @@ public abstract class PorticusMission { protected Runnable runnable; protected String[] command; protected long timeout = TimeUnit.SECONDS.toMillis(10); - private long start; + private volatile long start; + private volatile boolean started; + LongSupplier timeSource = System::currentTimeMillis; public PorticusMission() { this(UUID.randomUUID()); @@ -38,7 +41,8 @@ public PorticusMission(UUID uid) { * 通讯任务是否超时 */ public boolean isTimeout() { - return start + timeout < System.currentTimeMillis(); + long startedAt = start; + return started && timeSource.getAsLong() - startedAt >= timeout; } /** @@ -46,11 +50,26 @@ public boolean isTimeout() { * * @param target 发送目标，根据服务端类型传入对应玩家对象，当 API 类型为 SERVER 时传入 ProxyPlayer 类型，为 CLIENT 时则传入 Player 类型。 */ - public void run(@NotNull Object target) { - if (consumer != null || runnable != null) { - Porticus.INSTANCE.getMissions().add(this); + public synchronized void run(@NotNull Object target) { + if (started) { + throw new IllegalStateException("Porticus missions can only be run once"); + } + boolean trackCompletion = consumer != null || runnable != null; + if (trackCompletion) { + synchronized (Porticus.INSTANCE.getMissions()) { + for (PorticusMission mission : Porticus.INSTANCE.getMissions()) { + if (mission.getUID().equals(uid)) { + throw new IllegalStateException("A Porticus mission with the same UID is already pending"); + } + } + this.start = timeSource.getAsLong(); + this.started = true; + Porticus.INSTANCE.getMissions().add(this); + } + } else { + this.start = timeSource.getAsLong(); + this.started = true; } - this.start = System.currentTimeMillis(); } /** diff --git a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bukkitside/MissionBukkit.java b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bukkitside/MissionBukkit.java index 88d135c0d..cbdfbe259 100644 --- a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bukkitside/MissionBukkit.java +++ b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bukkitside/MissionBukkit.java @@ -10,7 +10,10 @@ import taboolib.module.porticus.common.MessageBuilder; import java.io.IOException; +import java.lang.reflect.Method; +import java.util.List; import java.util.UUID; +import java.util.function.Consumer; /** * Porticus @@ -32,23 +35,97 @@ public MissionBukkit(UUID uid) { @Override public void run(@NotNull Object target) { - super.run(target); - if (target instanceof Player) { - sendBukkitMessage((Player) target, command); - } else { + if (!(target instanceof Player)) { throw new IllegalStateException("target must be Player"); } + if (command == null) { + throw new IllegalStateException("command must be set before running mission"); + } + List messages; + try { + messages = MessageBuilder.create(command); + } catch (IOException e) { + throw new IllegalStateException("failed to encode mission command", e); + } + boolean tracked = consumer != null || runnable != null; + super.run(target); + try { + scheduleBukkitMessage((Player) target, messages, tracked); + } catch (Throwable t) { + Porticus.INSTANCE.getMissions().remove(this); + throw new IllegalStateException("failed to schedule mission message", t); + } } public void sendBukkitMessage(Player player, String[] command) { - Bukkit.getScheduler().runTaskAsynchronously(plugin, () -> { + try { + if (player == null) { + throw new IllegalArgumentException("player cannot be null"); + } + scheduleBukkitMessage(player, MessageBuilder.create(command), false); + } catch (Throwable t) { + t.printStackTrace(); + } + } + + private void scheduleBukkitMessage(Player player, List messages, boolean tracked) throws Exception { + Runnable failure = tracked ? () -> Porticus.INSTANCE.getMissions().remove(this) : () -> { + }; + Runnable sendTask = () -> { + if (tracked && !Porticus.INSTANCE.getMissions().contains(this)) { + return; + } try { - for (byte[] bytes : MessageBuilder.create(command)) { + for (byte[] bytes : messages) { + if (tracked && !Porticus.INSTANCE.getMissions().contains(this)) { + return; + } player.sendPluginMessage(plugin, Porticus.INSTANCE.getChannelId(), bytes); } - } catch (IOException e) { - e.printStackTrace(); + } catch (Throwable t) { + failure.run(); + t.printStackTrace(); + } + }; + if (isFolia()) { + runOnEntityScheduler(player, sendTask, failure); + } else if (Bukkit.isPrimaryThread()) { + sendTask.run(); + } else { + Bukkit.getScheduler().runTask(plugin, sendTask); + } + } + + private static boolean isFolia() { + try { + Class.forName("io.papermc.paper.threadedregions.RegionizedServer", false, playerClassLoader()); + return true; + } catch (Throwable ignored) { + return false; + } + } + + private static ClassLoader playerClassLoader() { + ClassLoader classLoader = Player.class.getClassLoader(); + return classLoader == null ? ClassLoader.getSystemClassLoader() : classLoader; + } + + private void runOnEntityScheduler(Player player, Runnable sendTask, Runnable retired) throws Exception { + Object scheduler = player.getClass().getMethod("getScheduler").invoke(player); + Method runMethod = null; + for (Method method : scheduler.getClass().getMethods()) { + if (method.getName().equals("run") && method.getParameterTypes().length == 3) { + runMethod = method; + break; } - }); + } + if (runMethod == null) { + throw new NoSuchMethodException("EntityScheduler#run"); + } + Consumer task = ignored -> sendTask.run(); + Object scheduled = runMethod.invoke(scheduler, plugin, task, retired); + if (scheduled == null) { + throw new IllegalStateException("EntityScheduler rejected Porticus message task"); + } } } diff --git a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bukkitside/PorticusListener.java b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bukkitside/PorticusListener.java index 17212d43c..58e9883c3 100644 --- a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bukkitside/PorticusListener.java +++ b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bukkitside/PorticusListener.java @@ -15,6 +15,9 @@ import taboolib.module.porticus.common.MessageReader; import java.io.IOException; +import java.lang.reflect.Method; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; /** * @author 坏黑 @@ -23,14 +26,17 @@ @SuppressWarnings("DuplicatedCode") public class PorticusListener implements Listener, PluginMessageListener { + private final Plugin plugin; + private final AtomicLong nextCacheWarning = new AtomicLong(); + public PorticusListener() { - Plugin plugin = JavaPlugin.getProvidingPlugin(Porticus.class); + plugin = JavaPlugin.getProvidingPlugin(Porticus.class); Bukkit.getPluginManager().registerEvents(this, plugin); Bukkit.getMessenger().registerIncomingPluginChannel(plugin, Porticus.INSTANCE.getChannelId(), this); Bukkit.getMessenger().registerOutgoingPluginChannel(plugin, Porticus.INSTANCE.getChannelId()); - Bukkit.getScheduler().runTaskTimer(plugin, () -> { + Runnable timeoutTask = () -> { for (PorticusMission mission : Porticus.INSTANCE.getMissions()) { - if (mission.isTimeout()) { + if (mission.isTimeout() && Porticus.INSTANCE.getMissions().remove(mission)) { if (mission.getTimeoutRunnable() != null) { try { mission.getTimeoutRunnable().run(); @@ -38,16 +44,21 @@ public PorticusListener() { t.printStackTrace(); } } - Porticus.INSTANCE.getMissions().remove(mission); } } - }, 0, 20); + MessageReader.cleanUp(); + }; + if (isFolia()) { + runGlobalTimer(plugin, timeoutTask); + } else { + Bukkit.getScheduler().runTaskTimer(plugin, timeoutTask, 0, 20); + } } @EventHandler(priority = EventPriority.MONITOR, ignoreCancelled = true) public void e(PorticusBukkitEvent e) { for (PorticusMission mission : Porticus.INSTANCE.getMissions()) { - if (mission.getUID().equals(e.getUID())) { + if (mission.getUID().equals(e.getUID()) && Porticus.INSTANCE.getMissions().remove(mission)) { if (mission.getResponseConsumer() != null) { try { mission.getResponseConsumer().accept(e.getArgs()); @@ -55,7 +66,7 @@ public void e(PorticusBukkitEvent e) { t.printStackTrace(); } } - Porticus.INSTANCE.getMissions().remove(mission); + break; } } } @@ -66,11 +77,57 @@ public void onPluginMessageReceived(@NotNull String channel, @NotNull Player pla try { Message message = MessageReader.read(bytes); if (message.isCompleted()) { - PorticusBukkitEvent.call(player, message.getMessages().get(0).getUID(), message.build()); + String[] args = message.buildOnce(); + if (args != null) { + PorticusBukkitEvent.call(player, message.getUID(), args); + } } + } catch (MessageReader.ProtocolException ignored) { + // Malformed or oversized plugin messages are rejected without flooding the server log. + } catch (MessageReader.CapacityException ex) { + warnCacheCapacity(ex); } catch (IOException ex) { ex.printStackTrace(); + } catch (Throwable t) { + t.printStackTrace(); + } + } + } + + private void warnCacheCapacity(IOException exception) { + long now = System.currentTimeMillis(); + long next = nextCacheWarning.get(); + if (now >= next && nextCacheWarning.compareAndSet(next, now + 10_000)) { + plugin.getLogger().warning("Porticus message cache rejected input: " + exception.getMessage()); + } + } + + private static boolean isFolia() { + try { + Class.forName("io.papermc.paper.threadedregions.RegionizedServer"); + return true; + } catch (Throwable ignored) { + return false; + } + } + + private static void runGlobalTimer(Plugin plugin, Runnable runnable) { + try { + Object scheduler = Bukkit.class.getMethod("getGlobalRegionScheduler").invoke(null); + Method runAtFixedRate = null; + for (Method method : scheduler.getClass().getMethods()) { + if (method.getName().equals("runAtFixedRate") && method.getParameterTypes().length == 4) { + runAtFixedRate = method; + break; + } + } + if (runAtFixedRate == null) { + throw new NoSuchMethodException("GlobalRegionScheduler#runAtFixedRate"); } + Consumer task = ignored -> runnable.run(); + runAtFixedRate.invoke(scheduler, plugin, task, 1L, 20L); + } catch (Throwable t) { + throw new IllegalStateException("Unable to schedule Porticus timeout task on Folia", t); } } } diff --git a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bungeeside/MissionBungee.java b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bungeeside/MissionBungee.java index df5ea82e9..9fbe04b68 100644 --- a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bungeeside/MissionBungee.java +++ b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bungeeside/MissionBungee.java @@ -5,12 +5,14 @@ import net.md_5.bungee.api.connection.ProxiedPlayer; import net.md_5.bungee.api.connection.Server; import net.md_5.bungee.api.plugin.Plugin; +import net.md_5.bungee.api.scheduler.ScheduledTask; import org.jetbrains.annotations.NotNull; import taboolib.module.porticus.Porticus; import taboolib.module.porticus.PorticusMission; import taboolib.module.porticus.common.MessageBuilder; import java.io.IOException; +import java.util.List; import java.util.UUID; /** @@ -22,8 +24,6 @@ */ public class MissionBungee extends PorticusMission { - private static final Plugin plugin = BungeeCord.getInstance().pluginManager.getPlugins().iterator().next(); - public MissionBungee() { super(); } @@ -34,35 +34,173 @@ public MissionBungee(UUID uid) { @Override public void run(@NotNull Object target) { + if (command == null) { + throw new IllegalStateException("command must be set before running mission"); + } + boolean tracked = consumer != null || runnable != null; + MessageTarget messageTarget = resolveTarget(target, tracked); + Plugin plugin = getPlugin(); + List messages; + try { + messages = MessageBuilder.create(command); + } catch (IOException e) { + throw new IllegalStateException("failed to encode mission command", e); + } super.run(target); - if (target instanceof Server) { - sendBungeeMessage((Server) target, command); - } else if (target instanceof ServerInfo) { - sendBungeeMessage((ServerInfo) target, command); - } else if (target instanceof ProxiedPlayer) { - sendBungeeMessage((ProxiedPlayer) target, command); - } else { - throw new IllegalStateException("target must be Server or ProxiedPlayer"); + try { + ScheduledTask task = BungeeCord.getInstance().getScheduler().runAsync(plugin, () -> { + if (tracked && !Porticus.INSTANCE.getMissions().contains(this)) { + return; + } + try { + sendMessages(messageTarget, messages, true); + } catch (Throwable t) { + Porticus.INSTANCE.getMissions().remove(this); + t.printStackTrace(); + } + }); + if (task == null) { + throw new IllegalStateException("Bungee scheduler rejected Porticus message task"); + } + } catch (Throwable t) { + Porticus.INSTANCE.getMissions().remove(this); + throw new IllegalStateException("failed to schedule mission message", t); } } public static void sendBungeeMessage(ProxiedPlayer player, String... args) { - sendBungeeMessage(player.getServer(), args); + sendStandalone(resolvePlayer(player), args); } public static void sendBungeeMessage(Server server, String... args) { - sendBungeeMessage(server.getInfo(), args); + sendStandalone(resolveServer(server), args); } public static void sendBungeeMessage(ServerInfo server, String... args) { - BungeeCord.getInstance().getScheduler().runAsync(plugin, () -> { - try { - for (byte[] bytes : MessageBuilder.create(args)) { - server.sendData(Porticus.INSTANCE.getChannelId(), bytes); + sendStandalone(resolveServerInfo(server), args); + } + + private static void sendStandalone(MessageTarget target, String[] args) { + try { + Plugin plugin = getPlugin(); + List messages = MessageBuilder.create(args); + ScheduledTask task = BungeeCord.getInstance().getScheduler().runAsync(plugin, () -> { + try { + sendMessages(target, messages, false); + } catch (Throwable t) { + t.printStackTrace(); } - } catch (IOException e) { - e.printStackTrace(); + }); + if (task == null) { + throw new IllegalStateException("Bungee scheduler rejected Porticus message task"); + } + } catch (Throwable t) { + t.printStackTrace(); + } + } + + private static void sendMessages(MessageTarget target, List messages, boolean mission) { + for (byte[] bytes : messages) { + if (mission && target instanceof MissionTarget && !((MissionTarget) target).missionPending()) { + return; } - }); + target.send(bytes); + } + } + + private MessageTarget resolveTarget(Object target, boolean tracked) { + MessageTarget resolved; + if (target instanceof Server) { + resolved = resolveServer((Server) target); + } else if (target instanceof ServerInfo) { + resolved = resolveServerInfo((ServerInfo) target); + } else if (target instanceof ProxiedPlayer) { + resolved = resolvePlayer((ProxiedPlayer) target); + } else { + throw new IllegalStateException("target must be Server, ServerInfo or ProxiedPlayer"); + } + return new MissionTarget(resolved, tracked); + } + + private static MessageTarget resolvePlayer(ProxiedPlayer player) { + if (player == null) { + throw new IllegalArgumentException("player cannot be null"); + } + Server connection = player.getServer(); + if (connection == null || !connection.isConnected()) { + throw new IllegalStateException("target player is not connected to a server"); + } + return bytes -> { + if (player.getServer() != connection || !connection.isConnected()) { + throw new IllegalStateException("target player changed server before Porticus message was sent"); + } + connection.sendData(Porticus.INSTANCE.getChannelId(), bytes); + }; + } + + private static MessageTarget resolveServer(Server server) { + if (server == null) { + throw new IllegalArgumentException("server cannot be null"); + } + if (server.getInfo() == null || !server.isConnected()) { + throw new IllegalStateException("target server connection is closed"); + } + return bytes -> { + if (!server.isConnected()) { + throw new IllegalStateException("target server connection is closed"); + } + server.sendData(Porticus.INSTANCE.getChannelId(), bytes); + }; + } + + private static MessageTarget resolveServerInfo(ServerInfo server) { + if (server == null) { + throw new IllegalArgumentException("server cannot be null"); + } + if (server.getPlayers().isEmpty()) { + throw new IllegalStateException("target server has no active player connection"); + } + return bytes -> { + if (!server.sendData(Porticus.INSTANCE.getChannelId(), bytes, false)) { + throw new IllegalStateException("target server has no active player connection"); + } + }; + } + + private static Plugin getPlugin() { + try { + Object instance = Class.forName("taboolib.platform.BungeePlugin").getMethod("getInstance").invoke(null); + if (instance instanceof Plugin) { + return (Plugin) instance; + } + } catch (Throwable t) { + throw new IllegalStateException("TabooLib BungeePlugin is not available", t); + } + throw new IllegalStateException("TabooLib BungeePlugin is not available"); + } + + private interface MessageTarget { + + void send(byte[] bytes); + } + + private final class MissionTarget implements MessageTarget { + + private final MessageTarget delegate; + private final boolean tracked; + + private MissionTarget(MessageTarget delegate, boolean tracked) { + this.delegate = delegate; + this.tracked = tracked; + } + + @Override + public void send(byte[] bytes) { + delegate.send(bytes); + } + + private boolean missionPending() { + return !tracked || Porticus.INSTANCE.getMissions().contains(MissionBungee.this); + } } } diff --git a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bungeeside/PorticusListener.java b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bungeeside/PorticusListener.java index 86fb2e5a4..871f38db4 100644 --- a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bungeeside/PorticusListener.java +++ b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bungeeside/PorticusListener.java @@ -16,6 +16,7 @@ import java.io.IOException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; /** * @author Bkm016 @@ -24,20 +25,26 @@ @SuppressWarnings("DuplicatedCode") public class PorticusListener implements Listener { - private static final Plugin plugin = BungeeCord.getInstance().pluginManager.getPlugins().iterator().next(); + private final Plugin plugin; + private final AtomicLong nextCacheWarning = new AtomicLong(); public PorticusListener() { + plugin = getPlugin(); ProxyServer.getInstance().registerChannel(Porticus.INSTANCE.getChannelId()); ProxyServer.getInstance().getPluginManager().registerListener(plugin, this); BungeeCord.getInstance().getScheduler().schedule(plugin, () -> { for (PorticusMission mission : Porticus.INSTANCE.getMissions()) { - if (!mission.isTimeout()) { + if (mission.isTimeout() && Porticus.INSTANCE.getMissions().remove(mission)) { if (mission.getTimeoutRunnable() != null) { - mission.getTimeoutRunnable().run(); + try { + mission.getTimeoutRunnable().run(); + } catch (Throwable t) { + t.printStackTrace(); + } } - Porticus.INSTANCE.getMissions().remove(mission); } } + MessageReader.cleanUp(); }, 1, 1, TimeUnit.SECONDS); } @@ -46,19 +53,41 @@ public void e(PorticusBungeeEvent e) { if (e.isCancelled()) { return; } - if (e.get(0).equals("porticus")) { - switch (e.get(1)) { + try { + for (PorticusMission mission : Porticus.INSTANCE.getMissions()) { + if (mission.getUID().equals(e.getUID()) && Porticus.INSTANCE.getMissions().remove(mission)) { + if (mission.getResponseConsumer() != null) { + try { + mission.getResponseConsumer().accept(e.getArgs()); + } catch (Throwable t) { + t.printStackTrace(); + } + } + return; + } + } + String[] args = e.getArgs(); + if (args.length < 2 || !"porticus".equals(args[0])) { + return; + } + switch (args[1]) { case "connect": { - ProxiedPlayer proxiedPlayer = ProxyServer.getInstance().getPlayer(e.get(2)); - ServerInfo serverInfo = ProxyServer.getInstance().getServerInfo(e.get(3)); + if (args.length < 4) { + return; + } + ProxiedPlayer proxiedPlayer = ProxyServer.getInstance().getPlayer(args[2]); + ServerInfo serverInfo = ProxyServer.getInstance().getServerInfo(args[3]); if (proxiedPlayer != null && serverInfo != null) { proxiedPlayer.connect(serverInfo); } break; } case "whois": { - ProxiedPlayer proxiedPlayer = ProxyServer.getInstance().getPlayer(e.get(2)); - if (proxiedPlayer != null) { + if (args.length < 3) { + return; + } + ProxiedPlayer proxiedPlayer = ProxyServer.getInstance().getPlayer(args[2]); + if (proxiedPlayer != null && proxiedPlayer.getServer() != null) { e.response(proxiedPlayer.getServer().getInfo().getName()); } break; @@ -66,19 +95,8 @@ public void e(PorticusBungeeEvent e) { default: break; } - } else { - for (PorticusMission mission : Porticus.INSTANCE.getMissions()) { - if (mission.getUID().equals(e.getUID())) { - if (mission.getResponseConsumer() != null) { - try { - mission.getResponseConsumer().accept(e.getArgs()); - } catch (Throwable t) { - t.printStackTrace(); - } - } - Porticus.INSTANCE.getMissions().remove(mission); - } - } + } catch (Throwable t) { + t.printStackTrace(); } } @@ -87,15 +105,44 @@ public void e(PluginMessageEvent e) { if (e.isCancelled()) { return; } - if (e.getSender() instanceof Server && e.getTag().equalsIgnoreCase(Porticus.INSTANCE.getChannelId())) { + if (e.getSender() instanceof Server && e.getReceiver() instanceof ProxiedPlayer && e.getTag().equalsIgnoreCase(Porticus.INSTANCE.getChannelId())) { try { Message message = MessageReader.read(e.getData()); if (message.isCompleted()) { - PorticusBungeeEvent.call((Server) e.getSender(), message.getMessages().get(0).getUID(), message.build()); + String[] args = message.buildOnce(); + if (args != null) { + PorticusBungeeEvent.call((Server) e.getSender(), message.getUID(), args); + } } + } catch (MessageReader.ProtocolException ignored) { + // Malformed or oversized plugin messages are rejected without flooding the proxy log. + } catch (MessageReader.CapacityException ex) { + warnCacheCapacity(ex); } catch (IOException ex) { ex.printStackTrace(); + } catch (Throwable t) { + t.printStackTrace(); + } + } + } + + private void warnCacheCapacity(IOException exception) { + long now = System.currentTimeMillis(); + long next = nextCacheWarning.get(); + if (now >= next && nextCacheWarning.compareAndSet(next, now + 10_000)) { + plugin.getLogger().warning("Porticus message cache rejected input: " + exception.getMessage()); + } + } + + private static Plugin getPlugin() { + try { + Object instance = Class.forName("taboolib.platform.BungeePlugin").getMethod("getInstance").invoke(null); + if (instance instanceof Plugin) { + return (Plugin) instance; } + } catch (Throwable t) { + throw new IllegalStateException("TabooLib BungeePlugin is not available", t); } + throw new IllegalStateException("TabooLib BungeePlugin is not available"); } } diff --git a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/ByteUtils.java b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/ByteUtils.java index c80c669d1..e89f76274 100644 --- a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/ByteUtils.java +++ b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/ByteUtils.java @@ -1,5 +1,8 @@ package taboolib.module.porticus.common; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CodingErrorAction; import java.nio.charset.StandardCharsets; import java.util.Base64; @@ -14,7 +17,16 @@ public static String serialize(String var) { } public static String deSerialize(String var) { - return new String(Base64.getDecoder().decode(var), StandardCharsets.UTF_8); + byte[] decoded = Base64.getDecoder().decode(var); + try { + return StandardCharsets.UTF_8.newDecoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT) + .decode(ByteBuffer.wrap(decoded)) + .toString(); + } catch (CharacterCodingException ex) { + throw new IllegalArgumentException("Serialized value is not valid UTF-8", ex); + } } public static String[] serialize(String... var) { @@ -28,7 +40,7 @@ public static String[] serialize(String... var) { public static String[] deSerialize(String... var) { String[] varEncode = new String[var.length]; for (int i = 0; i < var.length; i++) { - varEncode[i] = new String(Base64.getDecoder().decode(var[i]), StandardCharsets.UTF_8); + varEncode[i] = deSerialize(var[i]); } return varEncode; } diff --git a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/Message.java b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/Message.java index be855bdae..9a9b37b75 100644 --- a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/Message.java +++ b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/Message.java @@ -2,11 +2,18 @@ import com.google.common.collect.Lists; import com.google.gson.JsonArray; +import com.google.gson.JsonElement; import com.google.gson.JsonParser; import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; +import java.util.AbstractList; import java.util.Comparator; +import java.util.HashSet; import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; /** * 通讯信息容器 @@ -17,32 +24,250 @@ public class Message { private final List messages = Lists.newCopyOnWriteArrayList(); + private final List exposedMessages = new AbstractList() { + @Override + public MessagePacket get(int index) { + return messages.get(index); + } + + @Override + public int size() { + return messages.size(); + } + + @Override + public MessagePacket set(int index, MessagePacket element) { + synchronized (Message.this) { + MessagePacket previous = messages.set(index, element); + invalidateDecodedArguments(); + return previous; + } + } + + @Override + public void add(int index, MessagePacket element) { + synchronized (Message.this) { + messages.add(index, element); + invalidateDecodedArguments(); + } + } + + @Override + public MessagePacket remove(int index) { + synchronized (Message.this) { + MessagePacket removed = messages.remove(index); + invalidateDecodedArguments(); + return removed; + } + } + + @Override + public void clear() { + synchronized (Message.this) { + if (!messages.isEmpty()) { + messages.clear(); + invalidateDecodedArguments(); + } + } + } + }; + private final AtomicBoolean built = new AtomicBoolean(); + private final long createdAt; + private volatile long lastAccess; + private volatile long completedAt; + private volatile String[] decodedArguments; + private long cachedBytes; + + public Message() { + this(System.nanoTime()); + } + + Message(long createdAt) { + this.createdAt = createdAt; + this.lastAccess = createdAt; + } /** * 构建为可读取的通讯内容 */ @NotNull public String[] build() { - StringBuilder builder = new StringBuilder(); - messages.sort(Comparator.comparingInt(MessagePacket::getIndex)); - messages.forEach(message -> builder.append(message.getData())); - JsonArray json = new JsonParser().parse(ByteUtils.deSerialize(builder.toString())).getAsJsonArray(); - String[] args = new String[json.size()]; - for (int i = 0; i < json.size(); i++) { - args[i] = json.get(i).getAsString(); + String[] arguments = decodedArguments; + if (arguments == null) { + synchronized (this) { + arguments = decodedArguments; + if (arguments == null) { + arguments = decodeArguments(); + decodedArguments = arguments; + } + } + } + return arguments.clone(); + } + + /** + * 在所有数据包接收完成后仅构建一次。 + * + * @return 首次完整构建的内容，尚未完成或已经构建时返回 null + */ + @Nullable + public String[] buildOnce() { + if (!isCompleted() || !built.compareAndSet(false, true)) { + return null; + } + try { + return build(); + } catch (RuntimeException ex) { + built.set(false); + throw ex; } - return args; } /** * 所有数据包是否接收完成 */ public boolean isCompleted() { - return !messages.isEmpty() && messages.size() == messages.get(0).getTotal(); + List snapshot = Lists.newArrayList(messages); + if (snapshot.isEmpty()) { + return false; + } + try { + validateCompleted(snapshot); + return true; + } catch (IllegalStateException ignored) { + return false; + } + } + + /** + * 获取消息 UID。 + * + * @return 尚未接收任何数据包时返回 null + */ + @Nullable + public UUID getUID() { + return messages.isEmpty() ? null : messages.get(0).getUID(); } + /** + * 获取实时数据包列表，保持旧版 API 的可修改语义。 + */ @NotNull public List getMessages() { - return messages; + return exposedMessages; + } + + synchronized boolean addPacket(MessagePacket packet, int packetBytes, long now, MessageReader.CacheState cache) { + for (MessagePacket message : messages) { + if (!message.getUID().equals(packet.getUID())) { + throw new IllegalArgumentException("Message UID is inconsistent"); + } + if (message.getTotal() != packet.getTotal()) { + throw new IllegalArgumentException("Message total is inconsistent"); + } + if (message.getIndex() == packet.getIndex()) { + if (message.getData().equals(packet.getData())) { + lastAccess = now; + return false; + } + throw new IllegalArgumentException("Message packet data conflicts with an existing index"); + } + } + if (cachedBytes + packetBytes > MessageReader.MAX_MESSAGE_SIZE) { + throw new IllegalArgumentException("Message exceeds protocol cache size limit"); + } + if (!cache.reserve(packetBytes)) { + throw MessageReader.cacheCapacityExceeded("Message cache byte capacity exceeded"); + } + boolean added = false; + try { + messages.add(packet); + cachedBytes += packetBytes; + lastAccess = now; + decodedArguments = null; + if (messages.size() == packet.getTotal()) { + completedAt = now; + } + added = true; + return true; + } finally { + if (!added) { + cache.release(packetBytes); + } + } + } + + void validatePayload() { + if (decodedArguments == null) { + build(); + } + } + + synchronized long releaseCachedBytes() { + long released = cachedBytes; + cachedBytes = 0; + return released; + } + + boolean isExpired(long now) { + long completed = completedAt; + return now - lastAccess >= MessageReader.IDLE_TIMEOUT_NANOS + || now - createdAt >= MessageReader.MAX_LIFETIME_NANOS + || completed != 0 && now - completed >= MessageReader.COMPLETED_RETENTION_NANOS; + } + + private String[] decodeArguments() { + List snapshot = Lists.newArrayList(messages); + validateCompleted(snapshot); + messages.sort(Comparator.comparingInt(MessagePacket::getIndex)); + snapshot = Lists.newArrayList(messages); + StringBuilder builder = new StringBuilder(); + for (MessagePacket message : snapshot) { + builder.append(message.getData()); + } + JsonElement element; + try { + element = new JsonParser().parse(ByteUtils.deSerialize(builder.toString())); + } catch (RuntimeException ex) { + throw new IllegalArgumentException("Message payload is not valid JSON", ex); + } + if (!element.isJsonArray()) { + throw new IllegalArgumentException("Message payload must be a JSON array"); + } + JsonArray json = element.getAsJsonArray(); + String[] args = new String[json.size()]; + for (int i = 0; i < json.size(); i++) { + JsonElement argument = json.get(i); + if (!argument.isJsonPrimitive() || !argument.getAsJsonPrimitive().isString()) { + throw new IllegalArgumentException("Message argument must be a string"); + } + args[i] = argument.getAsString(); + } + return args; + } + + private void invalidateDecodedArguments() { + decodedArguments = null; + completedAt = 0; + } + + private static void validateCompleted(List packets) { + if (packets.isEmpty()) { + throw new IllegalStateException("Message is incomplete"); + } + MessagePacket first = packets.get(0); + int total = first.getTotal(); + if (packets.size() != total) { + throw new IllegalStateException("Message is incomplete"); + } + Set indexes = new HashSet<>(); + for (MessagePacket packet : packets) { + if (!first.getUID().equals(packet.getUID()) || packet.getTotal() != total) { + throw new IllegalStateException("Message metadata is inconsistent"); + } + if (packet.getIndex() < 1 || packet.getIndex() > total || !indexes.add(packet.getIndex())) { + throw new IllegalStateException("Message indexes are invalid"); + } + } } } diff --git a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessageBuilder.java b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessageBuilder.java index 49fbdec38..340314b94 100644 --- a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessageBuilder.java +++ b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessageBuilder.java @@ -8,6 +8,7 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.List; +import java.util.UUID; /** * 通讯信息数据包创建工具 @@ -29,25 +30,49 @@ public class MessageBuilder { * @param message 源数据 */ public static List create(String[] message) throws IOException { + if (message == null || message.length == 0 || message[0] == null) { + throw new IOException("Message UID is required"); + } + UUID uid; + try { + uid = UUID.fromString(message[0]); + } catch (IllegalArgumentException ex) { + throw new IOException("Message UID is invalid", ex); + } + if (!uid.toString().equalsIgnoreCase(message[0])) { + throw new IOException("Message UID is invalid"); + } List messages = Lists.newArrayList(); JsonArray array = new JsonArray(); for (int i = 1; i < message.length; i++) { + if (message[i] == null) { + throw new IOException("Message arguments cannot be null"); + } array.add(new JsonPrimitive(message[i])); } String source = ByteUtils.serialize(array.toString()); - int times = (int) Math.ceil(source.length() / (double) MESSAGE_LENGTH); + int times = (source.length() + MESSAGE_LENGTH - 1) / MESSAGE_LENGTH; + if (times < 1 || times > MessageReader.MAX_TOTAL) { + throw new IOException("Message contains too many packets"); + } + long totalBytes = 0; for (int i = 0; i < times; i++) { + int from = i * MESSAGE_LENGTH; + int to = Math.min(from + MESSAGE_LENGTH, source.length()); JsonObject json = new JsonObject(); - json.addProperty("uid", message[0]); + json.addProperty("uid", uid.toString()); json.addProperty("index", i + 1); json.addProperty("total", times); - if (source.length() < MESSAGE_LENGTH) { - json.addProperty("data", source); - } else { - json.addProperty("data", source.substring(0, source.length() - (source.length() - MESSAGE_LENGTH))); - source = source.substring(MESSAGE_LENGTH); + json.addProperty("data", source.substring(from, to)); + byte[] packet = json.toString().getBytes(StandardCharsets.UTF_8); + if (packet.length > MessageReader.MAX_PACKET_SIZE) { + throw new IOException("Message packet exceeds protocol size limit"); + } + totalBytes += packet.length; + if (totalBytes > MessageReader.MAX_MESSAGE_SIZE) { + throw new IOException("Message exceeds protocol cache size limit"); } - messages.add(json.toString().getBytes(StandardCharsets.UTF_8)); + messages.add(packet); } return messages; } diff --git a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessagePacket.java b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessagePacket.java index d22ca5acd..2f365cdd2 100644 --- a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessagePacket.java +++ b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessagePacket.java @@ -26,6 +26,18 @@ public class MessagePacket { private final int total; MessagePacket(UUID uid, String data, int index, int total) { + if (uid == null) { + throw new IllegalArgumentException("Message UID is required"); + } + if (data == null) { + throw new IllegalArgumentException("Message data is required"); + } + if (total < 1 || total > MessageReader.MAX_TOTAL) { + throw new IllegalArgumentException("Message total is out of range"); + } + if (index < 1 || index > total) { + throw new IllegalArgumentException("Message index is out of range"); + } this.uid = uid; this.data = data; this.index = index; diff --git a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessageReader.java b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessageReader.java index 09505ed8a..bbc19cafc 100644 --- a/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessageReader.java +++ b/module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/common/MessageReader.java @@ -1,14 +1,22 @@ package taboolib.module.porticus.common; -import com.google.common.cache.Cache; -import com.google.common.cache.CacheBuilder; +import com.google.gson.JsonElement; import com.google.gson.JsonObject; import com.google.gson.JsonParser; import java.io.IOException; +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CodingErrorAction; import java.nio.charset.StandardCharsets; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; /** * 通讯信息数据包读取工具 @@ -18,9 +26,45 @@ */ public class MessageReader { - private static final Cache queueMessages = CacheBuilder.newBuilder() - .expireAfterWrite(10, TimeUnit.SECONDS) - .build(); + static final int MAX_PACKET_SIZE = 32767; + static final int MAX_TOTAL = 1024; + static final int MAX_CACHED_MESSAGES = 1024; + static final long MAX_MESSAGE_SIZE = 4L * 1024 * 1024; + static final long MAX_CACHED_BYTES = 16L * 1024 * 1024; + static final long IDLE_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(10); + static final long COMPLETED_RETENTION_NANOS = TimeUnit.SECONDS.toNanos(10); + static final long MAX_LIFETIME_NANOS = TimeUnit.SECONDS.toNanos(30); + private static final long CLEANUP_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(1); + + private static final AtomicReference cache = new AtomicReference<>(new CacheState()); + + /** + * 清空消息缓存，并允许后续消息进入新的缓存状态。 + */ + public static void clear() { + replaceCache(true); + } + + /** + * 打开消息接收缓存。 + */ + public static void open() { + replaceCache(true); + } + + /** + * 关闭并清空消息接收缓存。 + */ + public static void close() { + replaceCache(false); + } + + /** + * 清理过期的未完成消息和已消费消息。 + */ + public static void cleanUp() { + cleanUp(cache.get(), System.nanoTime()); + } /** * 将通讯数据读取为数据包 @@ -28,7 +72,26 @@ public class MessageReader { * @param packet 通讯数据（未经过处理的原始内容） */ public static Message read(byte[] packet) throws IOException { - return read(new String(packet, StandardCharsets.UTF_8)); + if (packet == null || packet.length == 0) { + throw new ProtocolException("Message packet is empty"); + } + if (packet.length > MAX_PACKET_SIZE) { + throw new ProtocolException("Message packet exceeds protocol size limit"); + } + try { + String decoded = StandardCharsets.UTF_8.newDecoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT) + .decode(ByteBuffer.wrap(packet)) + .toString(); + return readValidated(decoded, packet.length, System.nanoTime()); + } catch (CharacterCodingException ex) { + throw new ProtocolException("Message packet is not valid UTF-8", ex); + } catch (CacheCapacityException ex) { + throw new CapacityException(ex.getMessage(), ex); + } catch (IllegalArgumentException ex) { + throw new ProtocolException("Invalid message packet", ex); + } } /** @@ -37,18 +100,312 @@ public static Message read(byte[] packet) throws IOException { * @param packet 通讯数据（未经过处理的原始内容） */ public static Message read(String packet) { - JsonObject json = new JsonParser().parse(packet).getAsJsonObject(); - Message message = queueMessages.getIfPresent(json.get("uid").getAsString()); - if (message == null) { - message = new Message(); - queueMessages.put(json.get("uid").getAsString(), message); - } - message.getMessages().add(new MessagePacket( - UUID.fromString(json.get("uid").getAsString()), - json.get("data").getAsString(), - json.get("index").getAsInt(), - json.get("total").getAsInt() - )); - return message; + return read(packet, System.nanoTime()); + } + + static Message read(String packet, long now) { + if (packet == null || packet.isEmpty()) { + throw new IllegalArgumentException("Message packet is empty"); + } + int packetBytes = packet.getBytes(StandardCharsets.UTF_8).length; + if (packetBytes > MAX_PACKET_SIZE) { + throw new IllegalArgumentException("Message packet exceeds protocol size limit"); + } + return readValidated(packet, packetBytes, now); + } + + private static Message readValidated(String source, int packetBytes, long now) { + ParsedPacket packet = parse(source); + while (true) { + CacheState state = cache.get(); + if (state.closed) { + throw new IllegalStateException("Message cache is closed"); + } + cleanUpIfNeeded(state, now); + String key = packet.uid.toString(); + Message message = computeMessage(state, key, packet, packetBytes, now, true); + if (state.closed || cache.get() != state) { + state.remove(key, message); + continue; + } + if (message.isCompleted()) { + try { + message.validatePayload(); + } catch (RuntimeException ex) { + state.remove(key, message); + throw ex; + } + } + if (state.closed || cache.get() != state) { + state.remove(key, message); + continue; + } + return message; + } + } + + private static Message computeMessage(CacheState state, String key, ParsedPacket packet, int packetBytes, long now, boolean retryAfterCleanup) { + AtomicReference deferredFailure = new AtomicReference<>(); + try { + Message message = state.messages.compute(key, (ignored, current) -> { + MessagePacket incoming = new MessagePacket(packet.uid, packet.data, packet.index, packet.total); + if (current != null && current.isExpired(now)) { + state.release(current.releaseCachedBytes()); + Message replacement = new Message(now); + try { + replacement.addPacket(incoming, packetBytes, now, state); + return replacement; + } catch (RuntimeException ex) { + state.slots.release(); + deferredFailure.set(ex); + return null; + } + } + if (current != null) { + current.addPacket(incoming, packetBytes, now, state); + return current; + } + if (!state.slots.tryAcquire()) { + throw cacheCapacityExceeded("Message cache entry capacity exceeded"); + } + Message created = new Message(now); + try { + created.addPacket(incoming, packetBytes, now, state); + return created; + } catch (RuntimeException ex) { + state.slots.release(); + throw ex; + } + }); + RuntimeException failure = deferredFailure.get(); + if (failure != null) { + throw failure; + } + return message; + } catch (CacheCapacityException ex) { + if (retryAfterCleanup && !state.closed) { + cleanUp(state, now); + return computeMessage(state, key, packet, packetBytes, now, false); + } + throw ex; + } + } + + static CacheCapacityException cacheCapacityExceeded(String message) { + return new CacheCapacityException(message); + } + + static void cleanUp(long now) { + cleanUp(cache.get(), now); + } + + static int cachedMessageCount() { + return cache.get().messages.size(); + } + + static long cachedByteCount() { + return cache.get().cachedBytes.get(); + } + + private static void cleanUpIfNeeded(CacheState state, long now) { + long next = state.nextCleanup.get(); + if (now >= next && state.nextCleanup.compareAndSet(next, now + CLEANUP_INTERVAL_NANOS)) { + cleanUp(state, now); + } + } + + private static void cleanUp(CacheState state, long now) { + for (String key : state.messages.keySet()) { + state.messages.computeIfPresent(key, (ignored, current) -> { + if (current.isExpired(now)) { + state.release(current.releaseCachedBytes()); + state.slots.release(); + return null; + } + return current; + }); + } + } + + private static void replaceCache(boolean keepAccepting) { + CacheState replacement = new CacheState(!keepAccepting); + CacheState previous = cache.getAndSet(replacement); + previous.closed = true; + previous.clear(); + } + + private static ParsedPacket parse(String source) { + JsonElement root; + try { + root = new JsonParser().parse(source); + } catch (RuntimeException ex) { + throw new IllegalArgumentException("Message packet is not valid JSON", ex); + } + if (!root.isJsonObject()) { + throw new IllegalArgumentException("Message packet must be a JSON object"); + } + JsonObject json = root.getAsJsonObject(); + String uidSource = stringField(json, "uid"); + String data = stringField(json, "data"); + int index = integerField(json, "index"); + int total = integerField(json, "total"); + if (total < 1 || total > MAX_TOTAL) { + throw new IllegalArgumentException("Message total is out of range"); + } + if (index < 1 || index > total) { + throw new IllegalArgumentException("Message index is out of range"); + } + UUID uid; + try { + uid = UUID.fromString(uidSource); + } catch (IllegalArgumentException ex) { + throw new IllegalArgumentException("Message UID is invalid", ex); + } + if (!uid.toString().equalsIgnoreCase(uidSource)) { + throw new IllegalArgumentException("Message UID is invalid"); + } + validateBase64Chunk(data, index, total); + return new ParsedPacket(uid, data, index, total); + } + + private static String stringField(JsonObject json, String name) { + JsonElement element = json.get(name); + if (element == null || !element.isJsonPrimitive() || !element.getAsJsonPrimitive().isString()) { + throw new IllegalArgumentException("Message field '" + name + "' must be a string"); + } + return element.getAsString(); + } + + private static int integerField(JsonObject json, String name) { + JsonElement element = json.get(name); + if (element == null || !element.isJsonPrimitive() || !element.getAsJsonPrimitive().isNumber()) { + throw new IllegalArgumentException("Message field '" + name + "' must be an integer"); + } + try { + return new BigDecimal(element.getAsString()).intValueExact(); + } catch (ArithmeticException | NumberFormatException ex) { + throw new IllegalArgumentException("Message field '" + name + "' must be an integer", ex); + } + } + + private static void validateBase64Chunk(String data, int index, int total) { + if (data.isEmpty()) { + throw new IllegalArgumentException("Message data is empty"); + } + if (data.length() > MessageBuilder.MESSAGE_LENGTH) { + throw new IllegalArgumentException("Message data exceeds packet chunk size limit"); + } + boolean padding = false; + int paddingLength = 0; + for (int i = 0; i < data.length(); i++) { + char character = data.charAt(i); + if (character == '=') { + if (index != total || ++paddingLength > 2) { + throw new IllegalArgumentException("Message data is not valid Base64"); + } + padding = true; + } else { + boolean base64 = character >= 'A' && character <= 'Z' + || character >= 'a' && character <= 'z' + || character >= '0' && character <= '9' + || character == '+' + || character == '/'; + if (!base64 || padding) { + throw new IllegalArgumentException("Message data is not valid Base64"); + } + } + } + } + + static final class CacheState { + + private final ConcurrentMap messages = new ConcurrentHashMap<>(); + private final Semaphore slots = new Semaphore(MAX_CACHED_MESSAGES); + private final AtomicLong cachedBytes = new AtomicLong(); + private final AtomicLong nextCleanup = new AtomicLong(); + private volatile boolean closed; + + private CacheState() { + this(false); + } + + private CacheState(boolean closed) { + this.closed = closed; + } + + boolean reserve(long bytes) { + while (true) { + long current = cachedBytes.get(); + if (bytes < 0 || current > MAX_CACHED_BYTES - bytes) { + return false; + } + if (cachedBytes.compareAndSet(current, current + bytes)) { + return true; + } + } + } + + void release(long bytes) { + if (bytes != 0) { + cachedBytes.addAndGet(-bytes); + } + } + + void remove(String key, Message message) { + if (messages.remove(key, message)) { + release(message.releaseCachedBytes()); + slots.release(); + } + } + + void clear() { + for (String key : messages.keySet()) { + messages.computeIfPresent(key, (ignored, current) -> { + release(current.releaseCachedBytes()); + slots.release(); + return null; + }); + } + } + } + + private static final class ParsedPacket { + + private final UUID uid; + private final String data; + private final int index; + private final int total; + + private ParsedPacket(UUID uid, String data, int index, int total) { + this.uid = uid; + this.data = data; + this.index = index; + this.total = total; + } + } + + private static final class CacheCapacityException extends IllegalStateException { + + private CacheCapacityException(String message) { + super(message); + } + } + + public static class ProtocolException extends IOException { + + public ProtocolException(String message) { + super(message); + } + + public ProtocolException(String message, Throwable cause) { + super(message, cause); + } + } + + public static class CapacityException extends IOException { + + public CapacityException(String message, Throwable cause) { + super(message, cause); + } } } diff --git a/module/minecraft/minecraft-porticus/src/main/kotlin/taboolib/module/porticus/Porticus.kt b/module/minecraft/minecraft-porticus/src/main/kotlin/taboolib/module/porticus/Porticus.kt index fc8b8d18f..e7c6912b5 100644 --- a/module/minecraft/minecraft-porticus/src/main/kotlin/taboolib/module/porticus/Porticus.kt +++ b/module/minecraft/minecraft-porticus/src/main/kotlin/taboolib/module/porticus/Porticus.kt @@ -10,6 +10,7 @@ import taboolib.common.platform.Platform import taboolib.common.platform.PlatformSide import taboolib.common.platform.function.pluginId import taboolib.common.util.unsafeLazy +import taboolib.module.porticus.common.MessageReader import java.util.concurrent.CopyOnWriteArrayList /** @@ -43,6 +44,7 @@ object Porticus { */ @Awake(LifeCycle.ENABLE) private fun onEnable() { + MessageReader.open() try { Bukkit.getServer() API = taboolib.module.porticus.bukkitside.PorticusAPI() @@ -54,4 +56,10 @@ object Porticus { } catch (ignored: Throwable) { } } + + @Awake(LifeCycle.DISABLE) + private fun onDisable() { + missions.clear() + MessageReader.close() + } } \ No newline at end of file diff --git a/module/minecraft/minecraft-porticus/src/test/java/taboolib/module/porticus/PorticusMissionTest.java b/module/minecraft/minecraft-porticus/src/test/java/taboolib/module/porticus/PorticusMissionTest.java new file mode 100644 index 000000000..f1489a744 --- /dev/null +++ b/module/minecraft/minecraft-porticus/src/test/java/taboolib/module/porticus/PorticusMissionTest.java @@ -0,0 +1,124 @@ +package taboolib.module.porticus; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class PorticusMissionTest { + + @AfterEach + void clearMissions() { + Porticus.INSTANCE.getMissions().clear(); + } + + @Test + void unstartedMissionNeverTimesOut() { + TestMission mission = new TestMission(); + mission.now = Long.MAX_VALUE; + mission.timeout(0, TimeUnit.MILLISECONDS); + + assertFalse(mission.isTimeout()); + } + + @Test + void timeoutUsesElapsedTimeAndIncludesBoundary() { + TestMission mission = new TestMission(); + mission.now = 1_000; + mission.timeout(100, TimeUnit.MILLISECONDS); + mission.run(new Object()); + + mission.now = 1_099; + assertFalse(mission.isTimeout()); + mission.now = 1_100; + assertTrue(mission.isTimeout()); + } + + @Test + void pendingMissionCannotBeStartedAgain() { + TestMission mission = new TestMission(); + mission.now = 1_000; + mission.onTimeout(() -> { + }); + mission.run(new Object()); + + mission.now = 2_000; + + assertThrows(IllegalStateException.class, () -> mission.run(new Object())); + assertEquals(1, Porticus.INSTANCE.getMissions().stream().filter(it -> it == mission).count()); + assertEquals(1_000, mission.getStart()); + } + + @Test + void differentMissionsCannotSharePendingUid() { + UUID uid = UUID.randomUUID(); + TestMission first = new TestMission(uid); + TestMission second = new TestMission(uid); + first.onTimeout(() -> { + }); + second.onTimeout(() -> { + }); + + first.run(new Object()); + + assertThrows(IllegalStateException.class, () -> second.run(new Object())); + assertEquals(1, Porticus.INSTANCE.getMissions().size()); + assertTrue(Porticus.INSTANCE.getMissions().contains(first)); + } + + @Test + void missionCanOnlyBeFinalizedOnce() { + TestMission mission = new TestMission(); + mission.now = 1_000; + mission.onTimeout(() -> { + }); + mission.run(new Object()); + + assertTrue(mission.cancel()); + assertFalse(mission.cancel()); + } + + @Test + void missionCannotBeReusedAfterFinalization() { + TestMission mission = new TestMission(); + mission.onTimeout(() -> { + }); + mission.now = 1_000; + mission.run(new Object()); + assertTrue(mission.cancel()); + + mission.now = 2_000; + + assertThrows(IllegalStateException.class, () -> mission.run(new Object())); + assertFalse(mission.pending()); + assertEquals(1_000, mission.getStart()); + } + + private static class TestMission extends PorticusMission { + + private long now; + + private TestMission() { + timeSource = () -> now; + } + + private TestMission(UUID uid) { + super(uid); + timeSource = () -> now; + } + + private boolean cancel() { + return Porticus.INSTANCE.getMissions().remove(this); + } + + private boolean pending() { + return Porticus.INSTANCE.getMissions().contains(this); + } + } +} diff --git a/module/minecraft/minecraft-porticus/src/test/java/taboolib/module/porticus/common/MessageProtocolTest.java b/module/minecraft/minecraft-porticus/src/test/java/taboolib/module/porticus/common/MessageProtocolTest.java new file mode 100644 index 000000000..94ec711af --- /dev/null +++ b/module/minecraft/minecraft-porticus/src/test/java/taboolib/module/porticus/common/MessageProtocolTest.java @@ -0,0 +1,331 @@ +package taboolib.module.porticus.common; + +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Base64; +import java.util.Collections; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class MessageProtocolTest { + + @BeforeEach + void resetCache() { + MessageReader.clear(); + } + + @AfterEach + void verifyCacheCanBeCleared() { + MessageReader.clear(); + assertEquals(0, MessageReader.cachedMessageCount()); + assertEquals(0, MessageReader.cachedByteCount()); + } + + @Test + void shouldRoundTripNormalMessage() throws IOException { + String uid = UUID.randomUUID().toString(); + String[] source = {uid, "command", "first", "second"}; + + Message message = readAll(MessageBuilder.create(source)); + + assertTrue(message.isCompleted()); + assertEquals(UUID.fromString(uid), message.getUID()); + assertArrayEquals(new String[]{"command", "first", "second"}, message.build()); + } + + @Test + void shouldRoundTripUnicodeMessage() throws IOException { + String[] source = {UUID.randomUUID().toString(), "你好，世界", "emoji: 😀", "日本語", "Привет"}; + + Message message = readAll(MessageBuilder.create(source)); + + assertArrayEquals(new String[]{"你好，世界", "emoji: 😀", "日本語", "Привет"}, message.build()); + } + + @Test + void shouldSplitAndReassembleMultiplePackets() throws IOException { + String large = repeat('a', MessageBuilder.MESSAGE_LENGTH * 2); + String[] source = {UUID.randomUUID().toString(), large, "tail"}; + List packets = MessageBuilder.create(source); + + assertTrue(packets.size() > 1); + for (byte[] packet : packets) { + assertTrue(packet.length <= MessageReader.MAX_PACKET_SIZE); + } + Message message = readAll(packets); + assertArrayEquals(new String[]{large, "tail"}, message.build()); + } + + @Test + void shouldReassembleOutOfOrderPackets() throws IOException { + String large = repeat('b', MessageBuilder.MESSAGE_LENGTH * 2); + List packets = new ArrayList<>(MessageBuilder.create(new String[]{UUID.randomUUID().toString(), large})); + Collections.reverse(packets); + + Message message = readAll(packets); + + assertTrue(message.isCompleted()); + assertArrayEquals(new String[]{large}, message.build()); + for (int i = 0; i < message.getMessages().size(); i++) { + assertEquals(i + 1, message.getMessages().get(i).getIndex()); + } + } + + @Test + void shouldDeduplicatePacketsByIndex() throws IOException { + String large = repeat('c', MessageBuilder.MESSAGE_LENGTH * 2); + List packets = MessageBuilder.create(new String[]{UUID.randomUUID().toString(), large}); + + Message message = MessageReader.read(packets.get(0)); + Message duplicate = MessageReader.read(packets.get(0)); + + assertEquals(1, duplicate.getMessages().size()); + assertEquals(message, duplicate); + for (int i = 1; i < packets.size(); i++) { + message = MessageReader.read(packets.get(i)); + } + assertTrue(message.isCompleted()); + assertArrayEquals(new String[]{large}, message.build()); + } + + @Test + void shouldRejectConflictingDataForTheSameIndex() throws IOException { + List packets = MessageBuilder.create(new String[]{UUID.randomUUID().toString(), repeat('c', MessageBuilder.MESSAGE_LENGTH * 2)}); + Message message = MessageReader.read(packets.get(0)); + JsonObject conflict = new JsonParser().parse(new String(packets.get(0), StandardCharsets.UTF_8)).getAsJsonObject(); + String data = conflict.get("data").getAsString(); + conflict.addProperty("data", (data.charAt(0) == 'A' ? 'B' : 'A') + data.substring(1)); + + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(conflict.toString())); + assertEquals(1, message.getMessages().size()); + } + + @Test + void shouldRemainIncompleteWhenPacketIsMissing() throws IOException { + String large = repeat('d', MessageBuilder.MESSAGE_LENGTH * 2); + List packets = MessageBuilder.create(new String[]{UUID.randomUUID().toString(), large}); + Message message = null; + + for (int i = 0; i < packets.size() - 1; i++) { + message = MessageReader.read(packets.get(i)); + } + + assertFalse(message.isCompleted()); + assertNull(message.buildOnce()); + Message incomplete = message; + assertThrows(IllegalStateException.class, incomplete::build); + } + + @Test + void shouldRejectIndexesAndTotalsOutsideProtocolBounds() { + String uid = UUID.randomUUID().toString(); + String data = ByteUtils.serialize("[]"); + + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(packet(uid, data, 0, 1))); + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(packet(uid, data, 2, 1))); + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(packet(uid, data, 1, 0))); + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(packet(uid, data, 1, MessageReader.MAX_TOTAL + 1))); + } + + @Test + void shouldRejectConflictingTotalWithoutMutatingCachedMessage() throws IOException { + String large = repeat('e', MessageBuilder.MESSAGE_LENGTH * 2); + List packets = MessageBuilder.create(new String[]{UUID.randomUUID().toString(), large}); + Message message = MessageReader.read(packets.get(0)); + JsonObject conflict = new JsonParser().parse(new String(packets.get(0), StandardCharsets.UTF_8)).getAsJsonObject(); + conflict.addProperty("total", conflict.get("total").getAsInt() + 1); + + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(conflict.toString())); + assertEquals(1, message.getMessages().size()); + for (int i = 1; i < packets.size(); i++) { + MessageReader.read(packets.get(i)); + } + assertTrue(message.isCompleted()); + assertArrayEquals(new String[]{large}, message.build()); + } + + @Test + void shouldRejectMalformedJsonBase64AndFieldTypesWithoutPollutingCache() throws IOException { + String uid = UUID.randomUUID().toString(); + + assertThrows(IllegalArgumentException.class, () -> MessageReader.read("not-json")); + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(packet(uid, "%%%", 1, 1))); + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(packet("1-1-1-1-1", ByteUtils.serialize("[]"), 1, 1))); + + JsonObject wrongType = new JsonObject(); + wrongType.addProperty("uid", uid); + wrongType.addProperty("data", ByteUtils.serialize("[]")); + wrongType.addProperty("index", "1"); + wrongType.addProperty("total", 1); + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(wrongType.toString())); + + Message valid = readAll(MessageBuilder.create(new String[]{uid, "valid"})); + assertTrue(valid.isCompleted()); + assertArrayEquals(new String[]{"valid"}, valid.build()); + } + + @Test + void shouldRejectInvalidOuterAndInnerUtf8() throws IOException { + assertThrows(IOException.class, () -> MessageReader.read(new byte[]{(byte) 0xC3, 0x28})); + + String uid = UUID.randomUUID().toString(); + byte[] invalidJsonBytes = new byte[]{'[', '"', (byte) 0xC3, '"', ']'}; + String encoded = Base64.getEncoder().encodeToString(invalidJsonBytes); + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(packet(uid, encoded, 1, 1))); + assertEquals(0, MessageReader.cachedMessageCount()); + + Message valid = readAll(MessageBuilder.create(new String[]{uid, "valid"})); + assertArrayEquals(new String[]{"valid"}, valid.build()); + } + + @Test + void shouldRejectOversizedRawPacketAndMessage() { + String oversized = repeat('x', MessageReader.MAX_PACKET_SIZE + 1); + + assertThrows(IllegalArgumentException.class, () -> MessageReader.read(oversized)); + assertThrows(IOException.class, () -> MessageReader.read(oversized.getBytes(StandardCharsets.UTF_8))); + assertThrows(IOException.class, () -> MessageBuilder.create(new String[]{ + UUID.randomUUID().toString(), + repeat('x', (int) MessageReader.MAX_MESSAGE_SIZE) + })); + } + + @Test + void shouldBuildCompletedMessageOnlyOnce() throws IOException { + String large = repeat('f', MessageBuilder.MESSAGE_LENGTH * 2); + Message message = readAll(MessageBuilder.create(new String[]{UUID.randomUUID().toString(), large, "done"})); + + assertArrayEquals(new String[]{large, "done"}, message.buildOnce()); + assertNull(message.buildOnce()); + assertTrue(message.isCompleted()); + assertArrayEquals(new String[]{large, "done"}, message.build()); + } + + @Test + void shouldSuppressCompletedMessageReplayUntilRetentionExpires() throws IOException { + List packets = MessageBuilder.create(new String[]{UUID.randomUUID().toString(), repeat('g', MessageBuilder.MESSAGE_LENGTH * 2)}); + long now = 1_000; + Message completed = readAll(packets, now); + assertArrayEquals(new String[]{repeat('g', MessageBuilder.MESSAGE_LENGTH * 2)}, completed.buildOnce()); + + Message replay = readAll(packets, now + 1); + + assertSame(completed, replay); + assertNull(replay.buildOnce()); + MessageReader.cleanUp(now + MessageReader.COMPLETED_RETENTION_NANOS + 1); + Message next = MessageReader.read(new String(packets.get(0), StandardCharsets.UTF_8), now + MessageReader.COMPLETED_RETENTION_NANOS + 2); + assertNotSame(completed, next); + assertFalse(next.isCompleted()); + } + + @Test + void shouldExpireIdlePartialMessageWithoutWaiting() throws IOException { + List packets = MessageBuilder.create(new String[]{UUID.randomUUID().toString(), repeat('h', MessageBuilder.MESSAGE_LENGTH * 2)}); + long now = 10_000; + Message partial = MessageReader.read(new String(packets.get(0), StandardCharsets.UTF_8), now); + + MessageReader.cleanUp(now + MessageReader.IDLE_TIMEOUT_NANOS + 1); + + assertEquals(0, MessageReader.cachedMessageCount()); + assertEquals(0, MessageReader.cachedByteCount()); + Message replacement = MessageReader.read(new String(packets.get(0), StandardCharsets.UTF_8), now + MessageReader.IDLE_TIMEOUT_NANOS + 2); + assertNotSame(partial, replacement); + } + + @Test + void shouldEnforcePerMessageAndEntryCacheCapacity() { + String uid = UUID.randomUUID().toString(); + String chunk = repeat('A', MessageBuilder.MESSAGE_LENGTH); + boolean rejected = false; + for (int index = 1; index <= 200; index++) { + try { + MessageReader.read(packet(uid, chunk, index, 200)); + } catch (IllegalArgumentException ex) { + rejected = true; + break; + } + } + assertTrue(rejected); + assertTrue(MessageReader.cachedByteCount() <= MessageReader.MAX_MESSAGE_SIZE); + + MessageReader.clear(); + String partialData = "Ww"; + for (int i = 0; i < MessageReader.MAX_CACHED_MESSAGES; i++) { + MessageReader.read(packet(UUID.randomUUID().toString(), partialData, 1, 2)); + } + assertEquals(MessageReader.MAX_CACHED_MESSAGES, MessageReader.cachedMessageCount()); + assertThrows(IllegalStateException.class, () -> MessageReader.read(packet(UUID.randomUUID().toString(), partialData, 1, 2))); + } + + @Test + void shouldRejectNewPacketsWhileCacheIsClosed() throws IOException { + String packet = new String(MessageBuilder.create(new String[]{UUID.randomUUID().toString(), "value"}).get(0), StandardCharsets.UTF_8); + + MessageReader.close(); + assertThrows(IllegalStateException.class, () -> MessageReader.read(packet)); + assertEquals(0, MessageReader.cachedMessageCount()); + + MessageReader.open(); + assertTrue(MessageReader.read(packet).isCompleted()); + } + + @Test + void shouldPreserveMutableLivePacketListCompatibility() throws IOException { + Message message = MessageReader.read(MessageBuilder.create(new String[]{UUID.randomUUID().toString(), "value"}).get(0)); + assertArrayEquals(new String[]{"value"}, message.build()); + + message.getMessages().clear(); + + assertFalse(message.isCompleted()); + assertThrows(IllegalStateException.class, message::build); + } + + private static Message readAll(List packets) throws IOException { + Message message = null; + for (byte[] packet : packets) { + message = MessageReader.read(packet); + } + return message; + } + + private static Message readAll(List packets, long now) { + Message message = null; + for (byte[] packet : packets) { + message = MessageReader.read(new String(packet, StandardCharsets.UTF_8), now); + } + return message; + } + + private static String packet(String uid, String data, int index, int total) { + JsonObject json = new JsonObject(); + json.addProperty("uid", uid); + json.addProperty("data", data); + json.addProperty("index", index); + json.addProperty("total", total); + return json.toString(); + } + + private static String repeat(char character, int length) { + StringBuilder builder = new StringBuilder(length); + for (int i = 0; i < length; i++) { + builder.append(character); + } + return builder.toString(); + } +}