diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt/MQTTProtocol.java b/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt/MQTTProtocol.java index ded52b254..37dcdf821 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt/MQTTProtocol.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt/MQTTProtocol.java @@ -152,6 +152,9 @@ public void connect(@NonNull @NotNull String sessionId, String username, String } public void connect(@NonNull @NotNull String sessionId, String username, String password, Publish willMsg) throws IOException { + if (password != null && username == null) { + throw new IOException("MQTT 3.1.1 password requires a username [MQTT-3.1.2-22]"); + } Connect connect = new Connect(); connect.setSessionId(sessionId); connectionSessionId = sessionId; @@ -159,6 +162,8 @@ public void connect(@NonNull @NotNull String sessionId, String username, String connect.setKeepAlive((int) keepAlive); if (username != null) { connect.setUsername(username); + } + if (password != null) { connect.setPassword(password.trim().toCharArray()); } if(willMsg != null) { diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt/packet/Connect.java b/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt/packet/Connect.java index d4402287f..9bd8f7dab 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt/packet/Connect.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt/packet/Connect.java @@ -27,6 +27,7 @@ import lombok.Setter; import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; /** * http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718028 @@ -244,6 +245,9 @@ public String toString() { } public int packFrame(Packet packet) { + if (passwordFlag && !usernameFlag) { + throw new IllegalArgumentException("MQTT 3.1.1 password requires a username [MQTT-3.1.2-22]"); + } ByteArrayOutputStream b = new ByteArrayOutputStream(); b.write(0); b.write(MQTT311.length); @@ -266,8 +270,11 @@ public int packFrame(Packet packet) { // Will never be greater then 2^31 int size = 10 + sessionId.length() + 2; - if (passwordFlag && usernameFlag) { - size += password.length + username.length() + 4; + if (usernameFlag) { + size += username.getBytes(StandardCharsets.UTF_8).length + 2; + } + if (passwordFlag) { + size += password.length + 2; } // Pack the header @@ -277,10 +284,14 @@ public int packFrame(Packet packet) { writeVariableInt(packet, size); packet.put(b.toByteArray()); writeUTF8(packet, sessionId); - writeUTF8(packet, username); - writeShort(packet, password.length); - for (char c : password) { - packet.put((byte) c); + if (usernameFlag) { + writeUTF8(packet, username); + } + if (passwordFlag) { + writeShort(packet, password.length); + for (char c : password) { + packet.put((byte) c); + } } return packet.position() - start; } diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt5/MQTT5Protocol.java b/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt5/MQTT5Protocol.java index 8f19503bc..90c1c97f3 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt5/MQTT5Protocol.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt5/MQTT5Protocol.java @@ -222,6 +222,8 @@ public void connect(@NonNull @NotNull String sessionId, String username, String connect.add(new ReceiveMaximum(serverReceiveMaximum)); if (username != null) { connect.setUsername(username); + } + if (password != null) { connect.setPassword(password.trim().toCharArray()); } if(willMsg != null){ diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt5/packet/Connect5.java b/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt5/packet/Connect5.java index c8b94d0d6..fd0fddd1c 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt5/packet/Connect5.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/mqtt5/packet/Connect5.java @@ -234,6 +234,8 @@ public String toString() { @Override public int packFrame(Packet packet) { + boolean usernamePresent = username != null; + boolean passwordPresent = password != null; // ---- Variable header ---- // Protocol Name "MQTT" int variableHeaderLength = 2 + mqtt.length; @@ -262,11 +264,11 @@ public int packFrame(Packet packet) { payloadLength += willSectionLength; } - if (hasUsername()) { + if (usernamePresent) { payloadLength += 2 + username.getBytes(java.nio.charset.StandardCharsets.UTF_8).length; } - if (hasPassword()) { + if (passwordPresent) { payloadLength += 2 + password.length; // binary with 2-byte length } @@ -292,10 +294,10 @@ public int packFrame(Packet packet) { connectFlag |= 0b00100000; } } - if (hasPassword()) { + if (passwordPresent) { connectFlag |= 0b01000000; } - if (hasUsername()) { + if (usernamePresent) { connectFlag |= (byte) 0b10000000; } packet.put(connectFlag); @@ -313,11 +315,11 @@ public int packFrame(Packet packet) { writeBuffer(willMsg, packet); } - if (hasUsername()) { + if (usernamePresent) { writeUTF8(packet, username); } - if (hasPassword()) { + if (passwordPresent) { byte[] passwordBytes = new byte[password.length]; for (int index = 0; index < password.length; index++) { passwordBytes[index] = (byte) password[index]; diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/mqtt/MQTTConnectCredentialsTest.java b/src/test/java/io/mapsmessaging/network/protocol/impl/mqtt/MQTTConnectCredentialsTest.java new file mode 100644 index 000000000..d303c5351 --- /dev/null +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/mqtt/MQTTConnectCredentialsTest.java @@ -0,0 +1,179 @@ +/* + * + * Copyright [ 2020 - 2024 ] Matthew Buckton + * Copyright [ 2024 - 2026 ] MapsMessaging B.V. + * + * Licensed under the Apache License, Version 2.0 with the Commons Clause + * (the "License"); you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * https://commonsclause.com/ + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.mapsmessaging.network.protocol.impl.mqtt; + +import io.mapsmessaging.config.protocol.impl.MqttConfig; +import io.mapsmessaging.dto.rest.config.network.EndPointConfigDTO; +import io.mapsmessaging.dto.rest.config.network.EndPointServerConfigDTO; +import io.mapsmessaging.network.io.EndPoint; +import io.mapsmessaging.network.io.EndPointServerStatus; +import io.mapsmessaging.network.io.EndPointStatus; +import io.mapsmessaging.network.io.Packet; +import io.mapsmessaging.network.io.ServerPacket; +import io.mapsmessaging.network.io.impl.SelectorTask; +import io.mapsmessaging.network.protocol.Protocol; +import io.mapsmessaging.network.protocol.impl.mqtt.packet.Connect; +import io.mapsmessaging.network.protocol.impl.mqtt.packet.MQTTPacket; +import io.mapsmessaging.network.protocol.impl.mqtt5.MQTT5Protocol; +import io.mapsmessaging.network.protocol.impl.mqtt5.packet.Connect5; +import java.io.IOException; +import java.lang.reflect.Field; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +class MQTTConnectCredentialsTest { + private static final AtomicInteger IDS = new AtomicInteger(); + + static Stream credentials() { + return Stream.of( + Arguments.of(false, null, null, 0), + Arguments.of(false, "user", null, 0x80), + Arguments.of(false, "user", "pass", 0xc0), + Arguments.of(false, "", null, 0x80), + Arguments.of(false, "", "", 0xc0), + Arguments.of(false, "üser", "pass", 0xc0), + Arguments.of(true, null, null, 0), + Arguments.of(true, "user", null, 0x80), + Arguments.of(true, null, "pass", 0x40), + Arguments.of(true, "user", "pass", 0xc0), + Arguments.of(true, "", null, 0x80), + Arguments.of(true, null, "", 0x40), + Arguments.of(true, "", "", 0xc0), + Arguments.of(true, " ", "pass", 0xc0)); + } + + @ParameterizedTest + @MethodSource("credentials") + void encoder_matches_credential_flags_and_payload(boolean mqtt5, String username, String password, int flags) throws Exception { + ServerPacket connect; + if (mqtt5) { + Connect5 frame = new Connect5(); + frame.setSessionId("client"); + frame.setUsername(username); + frame.setPassword(password == null ? null : password.toCharArray()); + connect = frame; + } else { + Connect frame = new Connect(); + frame.setSessionId("client"); + frame.setUsername(username); + frame.setPassword(password == null ? null : password.toCharArray()); + connect = frame; + } + assertCredentials(connect, mqtt5, username, password, flags); + } + + @ParameterizedTest + @MethodSource("credentials") + void outbound_protocol_emits_optional_credentials(boolean mqtt5, String username, String password, int flags) throws Exception { + Fixture fixture = fixture(mqtt5); + try { + fixture.protocol.connect("client", username, password); + assertEquals(1, fixture.frames.size()); + assertCredentials(fixture.frames.getFirst(), mqtt5, username, password, flags); + } finally { + fixture.protocol.setConnected(false); + fixture.protocol.close(); + } + } + + @Test + void mqtt311_encoder_rejects_password_without_username_before_writing() { + Connect frame = new Connect(); + frame.setSessionId("client"); + frame.setPassword("pass".toCharArray()); + Packet packet = new Packet(ByteBuffer.allocate(256)); + assertThrows(IllegalArgumentException.class, () -> frame.packFrame(packet)); + assertEquals(0, packet.position()); + } + + @Test + void mqtt311_protocol_rejects_password_without_username_before_sending() throws Exception { + Fixture fixture = fixture(false); + try { + assertThrows(IOException.class, () -> fixture.protocol.connect("client", null, "pass")); + assertTrue(fixture.frames.isEmpty()); + } finally { + fixture.protocol.setConnected(false); + fixture.protocol.close(); + } + } + + private static void assertCredentials(ServerPacket connect, boolean mqtt5, String username, String password, int flags) throws Exception { + Packet packet = new Packet(ByteBuffer.allocate(1024)); + connect.packFrame(packet); + packet.flip(); + byte fixed = packet.get(); + assertEquals(0x10, fixed & 0xff); + long remaining = MQTTPacket.readVariableInt(packet); + assertEquals(packet.available(), remaining, "Remaining Length must cover exactly the emitted bytes"); + int bodyStart = packet.position(); + assertEquals(flags, packet.getRawBuffer().get(bodyStart + 7) & 0xc0); + if (mqtt5) { + Connect5 decoded = new Connect5(fixed, remaining, packet); + assertEquals("client", decoded.getSessionId()); + assertEquals(username, decoded.getUsername()); + assertArrayEquals(password == null ? null : password.toCharArray(), decoded.getPassword()); + } else { + Connect decoded = new Connect(fixed, remaining, packet); + assertEquals("client", decoded.getSessionId()); + assertEquals(username, decoded.getUsername()); + assertArrayEquals(password == null ? null : password.toCharArray(), decoded.getPassword()); + } + assertFalse(packet.hasRemaining(), "No trailing credential bytes"); + } + + private static Fixture fixture(boolean mqtt5) throws Exception { + EndPoint endpoint = mock(EndPoint.class); + EndPointServerConfigDTO config = mock(EndPointServerConfigDTO.class); + MqttConfig mqttConfig = new MqttConfig(); + mqttConfig.setMaxServerKeepAlive(60); + when(config.getProtocolConfig("mqtt")).thenReturn(mqttConfig); + when(config.getEndPointConfig()).thenReturn(new EndPointConfigDTO("tcp")); + when(endpoint.getConfig()).thenReturn(config); + int id = IDS.incrementAndGet(); + when(endpoint.getName()).thenReturn("credentials-" + id); + when(endpoint.getJMXTypePath()).thenReturn(List.of("Credentials=" + id)); + when(endpoint.getEndPointStatus()).thenReturn(mock(EndPointStatus.class)); + when(endpoint.getServer()).thenReturn(mock(EndPointServerStatus.class)); + Protocol protocol = mqtt5 ? spy(new MQTT5Protocol(endpoint)) : spy(new MQTTProtocol(endpoint)); + Field selector = protocol.getClass().getDeclaredField("selectorTask"); + selector.setAccessible(true); + selector.set(protocol, mock(SelectorTask.class)); + List frames = new ArrayList<>(); + if (mqtt5) { + doAnswer(call -> { frames.add(call.getArgument(0)); return null; }).when((MQTT5Protocol) protocol).writeFrame(any(ServerPacket.class)); + } else { + doAnswer(call -> { frames.add(call.getArgument(0)); return null; }).when((MQTTProtocol) protocol).writeFrame(any(ServerPacket.class)); + } + return new Fixture(protocol, frames); + } + + private record Fixture(Protocol protocol, List frames) {} +}