Integration tests for MQTT topic parameters with guarded identity

runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v4/PublishIT.java

@@ -280,6 +280,26 @@ public void shouldPublish10k() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("server.when.topic.params.yaml")
+    @Specification({
+        "${net}/publish.topic.guarded.identity.param/client",
+        "${app}/publish.topic.guarded.identity.param/server"})
+    public void shouldPublishToTopicWithGuardedIdentityParam() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("server.when.topic.params.yaml")
+    @Specification({
+        "${net}/publish.invalid.topic.guarded.identity.param/client",
+        "${app}/publish.invalid.topic.guarded.identity.param/server"})
+    public void shouldRejectTopicWithInvalidGuardedIdentityParam() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Before
     public void setSubscriptionId()
     {

runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v4/SubscribeIT.java

@@ -365,6 +365,26 @@ public void shouldReplayIncompleteQoS2MessageAtReconnect() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("server.when.topic.params.yaml")
+    @Specification({
+        "${net}/subscribe.topic.guarded.identity.param/client",
+        "${app}/subscribe.topic.guarded.identity.param/server"})
+    public void shouldFilterGuardedIdentityParam() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("server.when.topic.params.yaml")
+    @Specification({
+        "${net}/subscribe.invalid.topic.guarded.identity.param/client",
+        "${app}/subscribe.invalid.topic.guarded.identity.param/server"})
+    public void shouldRejectInvalidGuardedIdentityParam() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Before
     public void setSubscriptionId()
     {

runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/PublishIT.java

@@ -513,4 +513,24 @@ public void shouldRejectLargeMessage() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Configuration("server.when.topic.params.yaml")
+    @Specification({
+        "${net}/publish.topic.guarded.identity.param/client",
+        "${app}/publish.topic.guarded.identity.param/server"})
+    public void shouldPublishToTopicWithGuardedIdentityParam() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("server.when.topic.params.yaml")
+    @Specification({
+        "${net}/publish.invalid.topic.guarded.identity.param/client",
+        "${app}/publish.invalid.topic.guarded.identity.param/server"})
+    public void shouldRejectTopicWithInvalidGuardedIdentityParam() throws Exception
+    {
+        k3po.finish();
+    }
 }

runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/SubscribeIT.java

@@ -528,6 +528,26 @@ public void shouldReplayIncompleteQoS2MessageAtReconnect() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("server.when.topic.params.yaml")
+    @Specification({
+        "${net}/subscribe.topic.guarded.identity.param/client",
+        "${app}/subscribe.topic.guarded.identity.param/server"})
+    public void shouldFilterGuardedIdentityParam() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("server.when.topic.params.yaml")
+    @Specification({
+        "${net}/subscribe.invalid.topic.guarded.identity.param/client",
+        "${app}/subscribe.invalid.topic.guarded.identity.param/server"})
+    public void shouldRejectInvalidGuardedIdentityParam() throws Exception
+    {
+        k3po.finish();
+    }
+
     private static int subscriptionId = 0;
     public static int supplySubscriptionId()
     {

specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/config/server.when.topic.params.yaml

@@ -0,0 +1,48 @@
+#
+# Copyright 2021-2024 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (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
+#
+# 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.
+#
+
+---
+name: test
+guards:
+  test0:
+    type: test
+    options:
+      credentials: TOKEN
+      lifetime: PT5S
+      challenge: PT5S
+bindings:
+  net0:
+    type: mqtt
+    kind: server
+    options:
+      authorization:
+        test0:
+          credentials:
+            connect:
+              username: Bearer {credentials}
+    routes:
+      - exit: app0
+        guarded:
+          test0: []
+        when:
+          - publish:
+            - topic: device/{id}/data
+              params:
+                id: ${guarded['test0'].identity}
+          - subscribe:
+              - topic: sensor/{id}/data
+                params:
+                  id: ${guarded['test0'].identity}

specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/publish.invalid.topic.guarded.identity.param/client.rpt

@@ -0,0 +1,46 @@
+#
+# Copyright 2021-2024 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (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
+#
+# 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.
+#
+
+connect "zilla://streams/app0"
+         option zilla:window 8192
+         option zilla:transmission "duplex"
+         option zilla:authorization 1L
+
+write zilla:begin.ext ${mqtt:beginEx()
+                            .typeId(zilla:id("mqtt"))
+                            .session()
+                              .flags("CLEAN_START")
+                              .clientId("client")
+                              .build()
+                            .build()}
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                              .typeId(zilla:id("mqtt"))
+                              .session()
+                                .flags("CLEAN_START")
+                                .subscribeQosMax(2)
+                                .publishQosMax(2)
+                                .packetSizeMax(66560)
+                                .capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS")
+                                .clientId("client")
+                                .build()
+                              .build()}
+
+connected
+
+read zilla:data.empty
+
+write abort

specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/publish.invalid.topic.guarded.identity.param/server.rpt

@@ -0,0 +1,49 @@
+#
+# Copyright 2021-2024 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (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
+#
+# 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.
+#
+
+accept "zilla://streams/app0"
+        option zilla:window 8192
+        option zilla:transmission "duplex"
+        option zilla:authorization 1L
+
+accepted
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                             .typeId(zilla:id("mqtt"))
+                             .session()
+                                .flags("CLEAN_START")
+                                .clientId("client")
+                                .build()
+                             .build()}
+
+write zilla:begin.ext ${mqtt:beginEx()
+                              .typeId(zilla:id("mqtt"))
+                              .session()
+                                .flags("CLEAN_START")
+                                .subscribeQosMax(2)
+                                .publishQosMax(2)
+                                .packetSizeMax(66560)
+                                .capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS")
+                                .clientId("client")
+                                .build()
+                              .build()}
+
+connected
+
+write zilla:data.empty
+write flush
+
+read aborted

specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/publish.topic.guarded.identity.param/client.rpt

@@ -0,0 +1,70 @@
+#
+# Copyright 2021-2024 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (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
+#
+# 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.
+#
+
+connect "zilla://streams/app0"
+         option zilla:window 8192
+         option zilla:transmission "duplex"
+         option zilla:authorization 1L
+
+write zilla:begin.ext ${mqtt:beginEx()
+                            .typeId(zilla:id("mqtt"))
+                            .session()
+                              .flags("CLEAN_START")
+                              .clientId("client")
+                              .build()
+                            .build()}
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                              .typeId(zilla:id("mqtt"))
+                              .session()
+                                .flags("CLEAN_START")
+                                .subscribeQosMax(2)
+                                .publishQosMax(2)
+                                .packetSizeMax(66560)
+                                .capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS")
+                                .clientId("client")
+                                .build()
+                              .build()}
+
+connected
+
+read zilla:data.empty
+read notify RECEIVED_SESSION_STATE
+
+
+connect await RECEIVED_SESSION_STATE
+        "zilla://streams/app0"
+         option zilla:window 8192
+         option zilla:transmission "duplex"
+         option zilla:authorization 1L
+
+write zilla:begin.ext ${mqtt:beginEx()
+                              .typeId(zilla:id("mqtt"))
+                              .publish()
+                                .clientId("client")
+                                .topic("device/test/data")
+                                .build()
+                              .build()}
+
+connected
+
+write zilla:data.ext ${mqtt:dataEx()
+                             .typeId(zilla:id("mqtt"))
+                             .publish()
+                                .build()
+                             .build()}
+
+write "message"

specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/publish.topic.guarded.identity.param/server.rpt

@@ -0,0 +1,67 @@
+#
+# Copyright 2021-2024 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (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
+#
+# 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.
+#
+
+accept "zilla://streams/app0"
+        option zilla:window 8192
+        option zilla:transmission "duplex"
+        option zilla:authorization 1L
+
+accepted
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                             .typeId(zilla:id("mqtt"))
+                             .session()
+                                .flags("CLEAN_START")
+                                .clientId("client")
+                                .build()
+                             .build()}
+
+write zilla:begin.ext ${mqtt:beginEx()
+                              .typeId(zilla:id("mqtt"))
+                              .session()
+                                .flags("CLEAN_START")
+                                .subscribeQosMax(2)
+                                .publishQosMax(2)
+                                .packetSizeMax(66560)
+                                .capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS")
+                                .clientId("client")
+                                .build()
+                              .build()}
+
+connected
+
+write zilla:data.empty
+write flush
+
+accepted
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                             .typeId(zilla:id("mqtt"))
+                             .publish()
+                                .clientId("client")
+                                .topic("device/test/data")
+                                .build()
+                             .build()}
+
+connected
+
+read zilla:data.ext ${mqtt:matchDataEx()
+                            .typeId(zilla:id("mqtt"))
+                            .publish()
+                                .build()
+                            .build()}
+
+read "message"