From bc37491328e7ef5216b839a9dc9a1ff8581cd706 Mon Sep 17 00:00:00 2001 From: Manevolent Date: Sat, 11 Aug 2018 18:44:44 -0400 Subject: [PATCH] Created a new test for fragments. If a packet fragment is received which wraps around the generation, it is lost. Also, packet fragmentation outgoing would cause a desync in the packet IDs. --- .../protocol/header/ClientPacketHeader.java | 16 ++ .../ts3j/protocol/header/PacketHeader.java | 2 + .../protocol/header/ServerPacketHeader.java | 15 + .../protocol/packet/PacketBody0Voice.java | 2 - .../protocol/packet/fragment/Fragments.java | 93 ++++++ .../{ => fragment}/PacketBodyFragment.java | 4 +- .../packet/fragment/PacketReassembly.java | 103 +++++++ .../client/LocalClientHandlerFull.java | 5 + .../client/AbstractTeamspeakClientSocket.java | 265 ++++++------------ .../manevolent/ts3j/CompressionTest.java | 3 +- .../github/manevolent/ts3j/FragmentTest.java | 61 ++++ .../manevolent/ts3j/ServerConnectionTest.java | 2 + 12 files changed, 391 insertions(+), 180 deletions(-) create mode 100644 src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/Fragments.java rename src/main/java/com/github/manevolent/ts3j/protocol/packet/{ => fragment}/PacketBodyFragment.java (81%) create mode 100644 src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java create mode 100644 src/test/java/com/github/manevolent/ts3j/FragmentTest.java diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/header/ClientPacketHeader.java b/src/main/java/com/github/manevolent/ts3j/protocol/header/ClientPacketHeader.java index 789955c..6672ba3 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/header/ClientPacketHeader.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/header/ClientPacketHeader.java @@ -56,4 +56,20 @@ public class ClientPacketHeader extends PacketHeader { public int getSize() { return super.getSize() + 2 + 1; } + + @Override + public PacketHeader clone() { + ClientPacketHeader header = new ClientPacketHeader(); + header.setPacketId(getPacketId()); + header.setType(getType()); + header.setClientId(getClientId()); + header.setGeneration(getGeneration()); + header.setPacketFlags(getPacketFlags()); + + byte[] mac = new byte[8]; + System.arraycopy(getMac(), 0, mac, 0, 8); + header.setMac(mac); + + return header; + } } \ No newline at end of file diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/header/PacketHeader.java b/src/main/java/com/github/manevolent/ts3j/protocol/header/PacketHeader.java index b677b5f..e2507e4 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/header/PacketHeader.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/header/PacketHeader.java @@ -110,4 +110,6 @@ public abstract class PacketHeader { if (packetGeneration < 0) throw new IllegalArgumentException("packetGeneration"); this.packetGeneration = packetGeneration; } + + public abstract PacketHeader clone(); } diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/header/ServerPacketHeader.java b/src/main/java/com/github/manevolent/ts3j/protocol/header/ServerPacketHeader.java index 3bfb742..7494ba6 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/header/ServerPacketHeader.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/header/ServerPacketHeader.java @@ -50,4 +50,19 @@ public class ServerPacketHeader extends PacketHeader { public int getSize() { return super.getSize() + 1; } + + @Override + public PacketHeader clone() { + ServerPacketHeader header = new ServerPacketHeader(); + header.setPacketId(getPacketId()); + header.setType(getType()); + header.setGeneration(getGeneration()); + header.setPacketFlags(getPacketFlags()); + + byte[] mac = new byte[8]; + System.arraycopy(getMac(), 0, mac, 0, 8); + header.setMac(mac); + + return header; + } } \ No newline at end of file diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBody0Voice.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBody0Voice.java index c620692..173241f 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBody0Voice.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBody0Voice.java @@ -53,8 +53,6 @@ public class PacketBody0Voice extends PacketBody { @Override public void read(ByteBuffer buffer) { - Ts3Debugging.debug(buffer.array()); - packetId = buffer.getShort() & 0x0000FFFF; if (getRole() == ProtocolRole.SERVER) diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/Fragments.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/Fragments.java new file mode 100644 index 0000000..22d2249 --- /dev/null +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/Fragments.java @@ -0,0 +1,93 @@ +package com.github.manevolent.ts3j.protocol.packet.fragment; + +import com.github.manevolent.ts3j.protocol.Packet; +import com.github.manevolent.ts3j.protocol.header.HeaderFlag; +import com.github.manevolent.ts3j.protocol.packet.PacketBodyCompressed; +import com.github.manevolent.ts3j.util.QuickLZ; +import com.github.manevolent.ts3j.util.Ts3Debugging; + +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.LinkedList; +import java.util.List; + +/** + * Packet fragmentation helper. Pulled out of the client logic for use in FragmentTest so we can track the stability + * of fragmentation. + */ +public final class Fragments { + /** + * Maximum segment size (header + body) allowed to leave the split method + */ + public static final int MAXIMUM_PACKET_SIZE = 500; + + /** + * Splits a packet into several sub-packets. + * @param packet Packet to send. + * @return List of packets to send + */ + public static List split(Packet packet) { + if (packet.getSize() > MAXIMUM_PACKET_SIZE) { // Check if packet is too large to send + + int size = packet.getBody().getSize(); + + ByteBuffer outputBuffer = ByteBuffer.allocate(size); + packet.getBody().write(outputBuffer); + + byte[] outputArray; + + if (packet.getHeader().getType().isCompressible()) { + byte[] compressed = QuickLZ.compress(outputBuffer.array(), 1); + + packet.getHeader().setPacketFlag(HeaderFlag.COMPRESSED, true); + packet.setBody(new PacketBodyCompressed(packet.getHeader().getType(), packet.getRole(), compressed)); + + if (packet.getSize() <= MAXIMUM_PACKET_SIZE) + return Collections.singletonList(packet); + + outputArray = compressed; + size = compressed.length; + } else + outputArray = outputBuffer.array(); + + List pieces = new LinkedList<>(); + + int maxFragmentSize = MAXIMUM_PACKET_SIZE - packet.getHeader().getSize(); + + for (int offs = 0; offs < size;) { + int flush = Math.min(maxFragmentSize, size - offs); + + boolean first = offs == 0; + boolean last = flush < maxFragmentSize; + + Packet piece = new Packet(packet.getRole()); + piece.setHeader(packet.getHeader().clone()); + + if (!first) // Only first packet has flags + piece.getHeader().setPacketFlags(HeaderFlag.NONE.getIndex()); + + // First and last packet get FRAGMENTED flag + piece.getHeader().setPacketFlag(HeaderFlag.FRAGMENTED, first || last); + + byte[] pieceBytes = new byte[flush]; + System.arraycopy(outputArray, offs, pieceBytes, 0, flush); + + piece.setBody(new PacketBodyFragment( + packet.getHeader().getType(), + packet.getHeader().getRole(), + pieceBytes + )); + + pieces.add(piece); + + offs += flush; + } + + return pieces; + } else { + // No splitting or compression + return Collections.singletonList(packet); + } + } + +} diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBodyFragment.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketBodyFragment.java similarity index 81% rename from src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBodyFragment.java rename to src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketBodyFragment.java index c5daccd..bbdf61f 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBodyFragment.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketBodyFragment.java @@ -1,6 +1,8 @@ -package com.github.manevolent.ts3j.protocol.packet; +package com.github.manevolent.ts3j.protocol.packet.fragment; import com.github.manevolent.ts3j.protocol.ProtocolRole; +import com.github.manevolent.ts3j.protocol.packet.PacketBody; +import com.github.manevolent.ts3j.protocol.packet.PacketBodyType; import java.nio.ByteBuffer; diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java new file mode 100644 index 0000000..b45f135 --- /dev/null +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java @@ -0,0 +1,103 @@ +package com.github.manevolent.ts3j.protocol.packet.fragment; + +import com.github.manevolent.ts3j.protocol.Packet; +import com.github.manevolent.ts3j.protocol.header.HeaderFlag; +import com.github.manevolent.ts3j.util.QuickLZ; +import com.github.manevolent.ts3j.util.Ts3Debugging; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.*; + +public class PacketReassembly { + private final Map queue = new LinkedHashMap<>(); + private boolean state = false; + + public Packet reassemble(Packet lastFragment) throws IOException { + // Pull out all other packets in the queue before this one which would also be fragmented + List reassemblyList = new ArrayList<>(); + for (int packetId = lastFragment.getHeader().getPacketId();; packetId--) { + if (packetId < 0) packetId = 65536 + packetId; + + Packet olderPacket = queue.get(packetId); + + if (olderPacket == null) + break; // Chain breaks here + else if (olderPacket.getHeader().getType() != lastFragment.getHeader().getType()) + continue; // skip + else { + reassemblyList.add(olderPacket); + } + } + + // Reverse the collection + Collections.reverse(reassemblyList); + + // Rebuild a master packet from the contents of all previous packet fragments + int totalLength = + reassemblyList + .stream() + .mapToInt(x -> x.getBody().getSize()) + .sum(); + + if (totalLength < 0) throw new IllegalArgumentException("reassembly too small: " + totalLength); + + Packet firstPacket = reassemblyList.get(0); + Packet reassembledPacket = new Packet(firstPacket.getRole()); + reassembledPacket.setHeader(firstPacket.getHeader()); + + ByteBuffer reassemblyBuffer = ByteBuffer.allocate(totalLength); + + reassembledPacket.getHeader().setPacketFlag(HeaderFlag.FRAGMENTED, false); + + for (Packet old : reassemblyList) + old.writeBody(reassemblyBuffer); + + if (firstPacket.getHeader().getPacketFlag(HeaderFlag.COMPRESSED)) + reassemblyBuffer = ByteBuffer.wrap(QuickLZ.decompress(reassemblyBuffer.array())); + + reassembledPacket.readBody(reassemblyBuffer); + + return reassembledPacket; + } + + public Packet put(Packet packet) throws IOException { + Packet reassembled; + + if (packet.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED)) { + boolean oldState = state; + + state = !state; + + if (!(packet.getBody() instanceof PacketBodyFragment)) + throw new IllegalArgumentException("packet fragment object is not representative of a fragment"); + + queue.put(packet.getHeader().getPacketId(), packet); + + if (oldState) { + reassembled = reassemble(packet); + } else { + reassembled = null; + } + } else { + if (state) { + queue.put(packet.getHeader().getPacketId(), packet); + + reassembled = null; + } else { + reassembled = packet; + } + } + + return reassembled; + } + + public boolean isReassembling() { + return state; + } + + public void reset() { + state = false; + queue.clear(); + } +} \ No newline at end of file diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerFull.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerFull.java index f00e152..0b8abcb 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerFull.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerFull.java @@ -27,6 +27,7 @@ public abstract class LocalClientHandlerFull extends LocalClientHandler { switch (packet.getBody().getType()) { case COMMAND: Ts3Debugging.debug(((PacketBody2Command) packet.getBody()).getText()); + try { handleCommand(MultiCommand.parse( packet.getRole(), @@ -35,8 +36,11 @@ public abstract class LocalClientHandlerFull extends LocalClientHandler { } catch (CommandProcessException e) { throw new IOException(e); } + break; case COMMAND_LOW: + Ts3Debugging.debug("[LOW] " + ((PacketBody2Command) packet.getBody()).getText()); + try { handleCommand(MultiCommand.parse( packet.getRole(), @@ -45,6 +49,7 @@ public abstract class LocalClientHandlerFull extends LocalClientHandler { } catch (CommandProcessException e) { throw new IOException(e); } + break; } } diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java index 8d93174..11827b5 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java @@ -11,6 +11,9 @@ import com.github.manevolent.ts3j.protocol.header.ClientPacketHeader; import com.github.manevolent.ts3j.protocol.header.HeaderFlag; import com.github.manevolent.ts3j.protocol.header.PacketHeader; import com.github.manevolent.ts3j.protocol.packet.*; +import com.github.manevolent.ts3j.protocol.packet.fragment.Fragments; +import com.github.manevolent.ts3j.protocol.packet.fragment.PacketBodyFragment; +import com.github.manevolent.ts3j.protocol.packet.fragment.PacketReassembly; import com.github.manevolent.ts3j.protocol.packet.handler.PacketHandler; import com.github.manevolent.ts3j.protocol.packet.handler.client.LocalClientHandler; import com.github.manevolent.ts3j.protocol.packet.statistics.PacketStatistics; @@ -52,7 +55,7 @@ public abstract class AbstractTeamspeakClientSocket private final LinkedBlockingDeque readQueue = new LinkedBlockingDeque<>(65536); - private final Map reassemblyQueue = new ConcurrentHashMap<>(); + private final Map reassemblyQueue = new ConcurrentHashMap<>(); private final Map pingQueue = new ConcurrentHashMap<>(); private final Map sendQueue = new ConcurrentHashMap<>(); @@ -61,6 +64,8 @@ public abstract class AbstractTeamspeakClientSocket private final Map localSendCounter = new ConcurrentHashMap<>(); private final Map remoteSendCounter = new ConcurrentHashMap<>(); + private final Map sendLocks = new HashMap<>(); + private CommandProcessor commandProcessor; private Consumer exceptionHandler = throwable -> { @@ -104,7 +109,7 @@ public abstract class AbstractTeamspeakClientSocket // Initialize reassembly queue for (PacketBodyType bodyType : PacketBodyType.values()) if (bodyType.isSplittable()) - reassemblyQueue.put(bodyType, new Reassembly()); + reassemblyQueue.put(bodyType, new PacketReassembly()); // Initialize counters for (PacketBodyType bodyType : PacketBodyType.values()) { @@ -115,6 +120,8 @@ public abstract class AbstractTeamspeakClientSocket localSendCounter.put(bodyType, new LocalCounterFull(65536, true)); remoteSendCounter.put(bodyType, new RemoteCounterFull(65536, 100)); } + + sendLocks.put(bodyType, new Object()); } // Initialize statistics @@ -254,11 +261,13 @@ public abstract class AbstractTeamspeakClientSocket @Override public void writePacket(PacketBody body) throws IOException, TimeoutException { - // Construct a network packet + // Construct a packet object Packet packet = new Packet(body.getRole()); + + // Set body packet.setBody(body); - // Generate header + // Generate new header (values get filled in as we go, and primarily in writePacketIntl) PacketHeader header = newOutgoingHeader(); // Ensure type matches @@ -270,7 +279,75 @@ public abstract class AbstractTeamspeakClientSocket if (header instanceof ClientPacketHeader) ((ClientPacketHeader) header).setClientId(clientId); + // Static header assignments (protocol expectations based on type) + switch (header.getType()) { + case INIT1: + case PONG: + case PING: + // setHeaderValues will do this, too + header.setPacketFlag(HeaderFlag.UNENCRYPTED, true); + break; + case COMMAND: + case COMMAND_LOW: + header.setPacketFlag(HeaderFlag.NEW_PROTOCOL, true); + break; + case VOICE: + // > X is a ushort in H2N order of an own audio packet counter + // it seems it can be the same as the packet counter so we will let the packethandler do it. + ((PacketBody0Voice)body).setPacketId(header.getPacketId()); + break; + } + + // Assign header pointer + packet.setHeader(header); + + // Lower-level write; helper is done + writePacket(packet); + } + + @Override + public void writePacket(Packet packet) throws IOException, TimeoutException { + // Sanity checks. People are insane! + if (packet.getHeader() == null) + throw new NullPointerException("header"); + else if (packet.getHeader().getType() == null) + throw new NullPointerException("header type"); + else if (packet.getHeader().getRole() != getRole().getOut()) + throw new IllegalArgumentException("packet role mismatch: " + + packet.getHeader().getRole().name() + " != " + + getRole().getOut().name()); + + if (packet.getHeader().getType().isSplittable()) { + // Split up the outgoing packet into valid pieces + // it's better to do this all at once instead of sending them as they are split, + // because we might get halfway through splitting them then throw an exception and + // we can't write the rest. + List packets = Fragments.split(packet); + + // Note, it's very important we send these packets synchronously. + // we allow the split to happen multi-thread, but we block until all + // chunks are sent synchronously. So, all chunks of packets are synchronous. + // this happens because we care that the packet IDs are contiguous. + synchronized (sendLocks.get(packet.getHeader().getType())) { + for (Packet piece : packets) + writePacketIntl(piece); // Write the individual piece, incrementing its packet + } + } else { + // Ensure the outbound packet's maximum size isn't too large if it can't be split (which it can't here) + if (packet.getSize() > Fragments.MAXIMUM_PACKET_SIZE) + throw new IllegalArgumentException("packet too large: " + packet.getSize() + "; " + + packet.getHeader().getType().name() + " cannot split"); + + // Don't lock here, for the slight performance benefit. + writePacketIntl(packet); + } + } + + private void writePacketIntl(Packet packet) throws IOException, TimeoutException { + PacketHeader header = packet.getHeader(); + // Count packet + // depends on the actual packet type coming in Pair counter; switch (header.getType()) { @@ -306,104 +383,20 @@ public abstract class AbstractTeamspeakClientSocket break; } + // Use the counters determined above to generate a packet id and generation id header.setPacketId(counter.getKey()); header.setGeneration(counter.getValue()); - switch (header.getType()) { - case INIT1: - case PONG: - case PING: - // setHeaderValues will do this, too - header.setPacketFlag(HeaderFlag.UNENCRYPTED, true); - break; - case COMMAND: - case COMMAND_LOW: - header.setPacketFlag(HeaderFlag.NEW_PROTOCOL, true); - break; - case VOICE: - // > X is a ushort in H2N order of an own audio packet counter - // it seems it can be the same as the packet counter so we will let the packethandler do it. - ((PacketBody0Voice)body).setPacketId(header.getPacketId()); - break; - } - - packet.setHeader(header); - - writePacket(packet); - } - - @Override - public void writePacket(Packet packet) throws IOException, TimeoutException { - if (packet.getHeader() == null) - throw new NullPointerException("header"); - else if (packet.getHeader().getRole() != getRole().getOut()) - throw new IllegalArgumentException("packet role mismatch: " + - packet.getHeader().getRole().name() + " != " + - getRole().getOut().name()); - - if (packet.getSize() > 500) { - int totalSize = packet.getBody().getSize(); - - ByteBuffer outputBuffer = ByteBuffer.allocate(totalSize); - packet.getBody().write(outputBuffer); - - if (packet.getHeader().getType().isCompressible()) { - byte[] compressed = QuickLZ.compress(outputBuffer.array(), 1); - packet.getHeader().setPacketFlag(HeaderFlag.COMPRESSED, true); - packet.setBody(new PacketBodyCompressed(packet.getHeader().getType(), packet.getRole(), compressed)); - - if (packet.getSize() <= 500) { - writePacketIntl(packet); - return; - } - } - - if (!packet.getHeader().getType().isSplittable()) - throw new IllegalArgumentException("packet too large: " + packet.getSize() + " (cannot split)"); - - for (int offs = 0; offs < totalSize;) { - int flush = Math.min(500 - packet.getHeader().getSize(), totalSize - offs); - boolean first = offs == 0; - boolean last = flush < 500; - - Packet piece = new Packet(packet.getRole()); - piece.setHeader(packet.getHeader()); - - if (!first) // Only first packet has flags - piece.getHeader().setPacketFlags(HeaderFlag.NONE.getIndex()); - - // First and last packet get FRAGMENTED flag - piece.getHeader().setPacketFlag(HeaderFlag.FRAGMENTED, first || last); - - byte[] pieceBytes = new byte[flush]; - System.arraycopy(outputBuffer.array(), offs, pieceBytes, 0, flush); - - piece.setBody(new PacketBodyFragment( - packet.getHeader().getType(), - packet.getHeader().getRole(), - pieceBytes - )); - - writePacketIntl(piece); - - offs += flush; - } - } else { - writePacketIntl(packet); - } - } - - private void writePacketIntl(Packet packet) throws IOException, TimeoutException { // Flush to a buffer ByteBuffer outputBuffer; - if (!packet.getHeader().getPacketFlag(HeaderFlag.UNENCRYPTED)) { + if (!packet.getHeader().getPacketFlag(HeaderFlag.UNENCRYPTED)) { // encrypt out Ts3Debugging.debug("[PROTOCOL] ENCRYPT " + packet.getHeader().getType().name() + " generation=" + packet.getHeader().getGeneration()); outputBuffer = getTransformation().encrypt(packet); - } else { + } else { // "fake encrypt" if (packet.getHeader().getType() == PacketBodyType.INIT1) { System.arraycopy(INIT1_MAC, 0, packet.getHeader().getMac(), 0, 8); } else { @@ -417,6 +410,7 @@ public abstract class AbstractTeamspeakClientSocket packet.writeBody(outputBuffer); } + // Construct actual network packet now that the protocol is finished. NetworkPacket networkPacket = new NetworkPacket( new DatagramPacket(outputBuffer.array(), outputBuffer.position()), packet.getHeader(), @@ -459,8 +453,10 @@ public abstract class AbstractTeamspeakClientSocket Ts3Debugging.debug("[PROTOCOL] WRITE " + networkPacket.getHeader().getType().name()); + // Real network send writeNetworkPacket(networkPacket); + // Now that the packet is sent, we need to track the stats PacketKind kind = networkPacket.getHeader().getType().getKind(); if (kind != null) packetStatistics @@ -656,6 +652,8 @@ public abstract class AbstractTeamspeakClientSocket pong.getHeader().setPacketFlag(HeaderFlag.UNENCRYPTED, true); pong.getHeader().setPacketId(packet.getHeader().getPacketId()); + pong.getHeader().setType(PacketBodyType.PONG); + pong.setBody(new PacketBody5Pong(getRole().getOut(), packet.getHeader().getPacketId())); writePacket(pong); @@ -1350,89 +1348,4 @@ public abstract class AbstractTeamspeakClientSocket } } - private class Reassembly { - private final Map queue = new LinkedHashMap<>(); - private boolean state = false; - - public Packet reassemble(Packet lastFragment) throws IOException { - // Pull out all other packets in the queue before this one which would also be fragmented - List reassemblyList = new ArrayList<>(); - for (int packetId = lastFragment.getHeader().getPacketId();; packetId--) { - Packet olderPacket = queue.get(packetId); - - if (olderPacket == null) - break; // Chain breaks here - else if (olderPacket.getHeader().getType() != lastFragment.getHeader().getType()) - continue; // skip - else { - reassemblyList.add(olderPacket); - } - } - - // Reverse the collection - Collections.reverse(reassemblyList); - - // Rebuild a master packet from the contents of all previous packet fragments - int totalLength = reassemblyList.stream().mapToInt(x -> x.getBody().getSize()).sum() + lastFragment.getBody().getSize(); - if (totalLength < 0) throw new IllegalArgumentException("reassembly too small: " + totalLength); - - Packet firstPacket = reassemblyList.get(0); - Packet reassembledPacket = new Packet(firstPacket.getRole()); - reassembledPacket.setHeader(firstPacket.getHeader()); - - ByteBuffer reassemblyBuffer = ByteBuffer.allocate(totalLength); - - reassembledPacket.getHeader().setPacketFlag(HeaderFlag.FRAGMENTED, false); - - for (Packet old : reassemblyList) - old.writeBody(reassemblyBuffer); - - if (firstPacket.getHeader().getPacketFlag(HeaderFlag.COMPRESSED)) - reassemblyBuffer = ByteBuffer.wrap(QuickLZ.decompress(reassemblyBuffer.array())); - - reassembledPacket.readBody(reassemblyBuffer); - - return reassembledPacket; - } - - public Packet put(Packet packet) throws IOException { - Packet reassembled; - - if (packet.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED)) { - boolean oldState = state; - - state = !state; - - if (!(packet.getBody() instanceof PacketBodyFragment)) - throw new IllegalArgumentException("packet fragment object is not representative of a fragment"); - - queue.put(packet.getHeader().getPacketId(), packet); - - if (oldState) { - reassembled = reassemble(packet); - } else { - reassembled = null; - } - } else { - if (state) { - queue.put(packet.getHeader().getPacketId(), packet); - - reassembled = null; - } else { - reassembled = packet; - } - } - - return reassembled; - } - - public boolean isReassembling() { - return state; - } - - public void reset() { - state = false; - queue.clear(); - } - } } diff --git a/src/test/java/com/github/manevolent/ts3j/CompressionTest.java b/src/test/java/com/github/manevolent/ts3j/CompressionTest.java index 310f61b..2836a70 100644 --- a/src/test/java/com/github/manevolent/ts3j/CompressionTest.java +++ b/src/test/java/com/github/manevolent/ts3j/CompressionTest.java @@ -2,6 +2,7 @@ package com.github.manevolent.ts3j; import com.github.manevolent.ts3j.util.QuickLZ; import junit.framework.TestCase; +import org.bouncycastle.util.encoders.Base64; import java.util.Random; @@ -28,6 +29,6 @@ public class CompressionTest extends TestCase { r.nextBytes(b); QuickLZ.decompress(QuickLZ.compress(b, 3)); } - } + } } diff --git a/src/test/java/com/github/manevolent/ts3j/FragmentTest.java b/src/test/java/com/github/manevolent/ts3j/FragmentTest.java new file mode 100644 index 0000000..acdb22a --- /dev/null +++ b/src/test/java/com/github/manevolent/ts3j/FragmentTest.java @@ -0,0 +1,61 @@ +package com.github.manevolent.ts3j; + +import com.github.manevolent.ts3j.protocol.Packet; +import com.github.manevolent.ts3j.protocol.ProtocolRole; +import com.github.manevolent.ts3j.protocol.header.ClientPacketHeader; +import com.github.manevolent.ts3j.protocol.packet.PacketBody2Command; +import com.github.manevolent.ts3j.protocol.packet.PacketBodyType; +import com.github.manevolent.ts3j.protocol.packet.fragment.Fragments; +import com.github.manevolent.ts3j.protocol.packet.fragment.PacketReassembly; +import com.github.manevolent.ts3j.protocol.socket.client.AbstractTeamspeakClientSocket; +import com.github.manevolent.ts3j.util.Pair; +import com.github.manevolent.ts3j.util.QuickLZ; +import com.github.manevolent.ts3j.util.Ts3Debugging; +import junit.framework.TestCase; +import org.bouncycastle.util.encoders.Base64; + +import java.io.IOException; +import java.util.List; +import java.util.Random; + +public class FragmentTest extends TestCase { + + public static void main(String[] args) throws Exception { + new FragmentTest().testParser(); + } + + public void testParser() throws Exception { + Ts3Debugging.setEnabled(true); + + byte[] data = new byte[2000]; + new Random(0x1234567).nextBytes(data); + + ClientPacketHeader header = new ClientPacketHeader(); + Packet largePacket = new Packet(ProtocolRole.CLIENT, header); + header.setType(PacketBodyType.COMMAND); + + String text = Base64.toBase64String(data); + largePacket.setBody(new PacketBody2Command(ProtocolRole.CLIENT, text)); + + PacketReassembly reassembly = new PacketReassembly(); + AbstractTeamspeakClientSocket.LocalCounterFull counter = + new AbstractTeamspeakClientSocket.LocalCounterFull(65536, true); + + counter.setPacketId(65534); // Test wrapping around the generation at the same time :) + + List pieces = Fragments.split(largePacket); + Ts3Debugging.debug("Created " + pieces.size() + " pieces (total=" + data.length + ")."); + + for (Packet piece : pieces) { + Pair packetId = counter.next(); + piece.getHeader().setPacketId(packetId.getKey()); + piece.getHeader().setGeneration(packetId.getValue()); + + Packet out = reassembly.put(piece); + if (out != null) + assertEquals(text, ((PacketBody2Command)out.getBody()).getText()); + } + + } + +} diff --git a/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java b/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java index 7414823..63a40ff 100644 --- a/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java +++ b/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java @@ -146,6 +146,8 @@ public class ServerConnectionTest { client.subscribeAll(); + for (int i = 0; i < 1000; i ++) client.getClientInfo(client.getClientId()); + Thread.sleep(100000000L); } catch (Throwable ex) { ex.printStackTrace();