Merge pull request #56 from MO-RISE/feature/seperate-mqtt-root-topics

Feature/seperate mqtt root topics

broker/acl.conf

@@ -1,6 +1,10 @@
+%% Allow access for internal users to PONTOS and PONTOS_HUB root topics
+{allow, {user, "__internal__"}, publish, ["PONTOS_HUB/#"]}.
+{allow, {user, "__internal__"}, subscribe, ["PONTOS/#"]}.
+
 %% Allow subscriptions to the PONTOS root topic from anyone
 %% that is already authenticated
-{allow, all, subscribe, ["PONTOS/#"]}.
+{allow, all, subscribe, ["PONTOS_HUB/#"]}.
 
 %% Deny everything else
 {deny, all}.

docker-compose.base.yml

@@ -50,7 +50,7 @@ services:
       # Do NOT allow for subscribers to upgrade QoS
       - EMQX_MQTT__UPGRADE_QOS=false
       # Rate limit the number of published messages per client on the ws listener
-      - EMQX_LISTENERS__WS__DEFAULT__MESSAGES_RATE="5000/s"
+      - EMQX_LISTENERS__WS__DEFAULT__MESSAGES_RATE="2000/s"
       # Lets disable the force_shutdown feature to avoid kicking out publishers with a high burst rate of messages
       - EMQX_FORCE_SHUTDOWN__ENABLE=false
       # Avoid queuing QoS0 messages
@@ -159,6 +159,7 @@ services:
     environment:
       - MQTT_BROKER_HOST=emqx
       - MQTT_BROKER_PORT=1883
+      - MQTT_USER=__internal__
       - MQTT_CLIENT_ID=
       - MQTT_CLEAN_START=True
       - MQTT_SUBSCRIBE_TOPIC=$$share/ingestors/PONTOS/+/+/+
@@ -185,6 +186,21 @@ services:
         max-size: "10m"
         max-file: "5"
 
+  mqtt-editor:
+    image: ghcr.io/mo-rise/porla-mqtt:v0.2.1
+    restart: unless-stopped
+    depends_on:
+      - emqx
+    deploy:
+      mode: replicated
+      replicas: 3
+    command:
+      [
+        "mqtt --host emqx --port 1883 --user __internal__ subscribe -t '$$share/filterers/PONTOS/+/+/+' --line '{topic} {message}'
+        | sed -u '/^$/d'
+        | sed -u 's/PONTOS/PONTOS_HUB/'
+        | mqtt --host emqx --port 1883 --user __internal__ publish --line '{topic} {message}'"
+      ]
 volumes:
   vol-emqx-data:
   vol-pg-data:

tests/10-test-base-setup.bats

@@ -10,7 +10,7 @@ load "./bats-helpers/bats-assert/load"
 setup_file() {
     docker pull hivemq/mqtt-cli:4.15.0
     docker compose -f docker-compose.base.yml up -d
-    sleep 10
+    sleep 15
 }
 
 teardown_file() {
@@ -68,4 +68,22 @@ teardown_file() {
     run curl --silent localhost/api/vessel_ids
     assert_equal "$status" 0
     assert_output --partial 'test_vessel'
+}
+
+@test "BASE: mqtt editor" {
+    # Start a subscriber in the background and let it run for 10s
+    docker run --name subscriber --detach --network='host' hivemq/mqtt-cli:4.15.0 sub -v -h localhost -p 80 -ws -ws:path mqtt -t PONTOS_HUB/#
+
+    sleep 2
+
+    # Publish an actual payload that should be rewritten by the mqtt editor
+    run docker run --network='host' hivemq/mqtt-cli:4.15.0 pub -v -h localhost -p 80 -ws -ws:path mqtt -t PONTOS/test_vessel/test_parameter/1 -m '{"timestamp": 12345678, "value": 42}'
+    assert_line --partial 'received PUBLISH acknowledgement'
+
+    sleep 1
+
+    run docker logs subscriber
+    assert_line --partial '{"timestamp": 12345678, "value": 42}'
+
+    docker kill subscriber
 }

tests/20-test-issue-jwt.bats

@@ -10,7 +10,7 @@ load "./bats-helpers/bats-assert/load"
 setup_file() {
     docker pull hivemq/mqtt-cli:4.15.0
     docker compose -f docker-compose.base.yml -f docker-compose.auth.yml -f tests/docker-compose.auth-test.yml up -d --build
-    sleep 10
+    sleep 15
 }
 
 teardown_file() {

tests/21-test-auth-setup.bats

@@ -10,7 +10,7 @@ load "./bats-helpers/bats-assert/load"
 setup_file() {
     docker pull hivemq/mqtt-cli:4.15.0
     docker compose -f docker-compose.base.yml -f docker-compose.auth.yml up -d --build
-    sleep 10
+    sleep 15
 }
 
 teardown_file() {
@@ -83,7 +83,7 @@ teardown_file() {
     assert_line --partial "failed SUBSCRIBE to TOPIC 'anything/anything/#': Server closed connection without DISCONNECT."
 
     # However, we should be allowed to subscribe to the PONTOS root topic
-    run timeout --preserve-status 5s docker run --network='host' hivemq/mqtt-cli:4.15.0 sub -v -h localhost -p 80 -u '__token__' -pw "$token" -ws -ws:path mqtt -t PONTOS/#
+    run timeout --preserve-status 5s docker run --network='host' hivemq/mqtt-cli:4.15.0 sub -v -h localhost -p 80 -u '__token__' -pw "$token" -ws -ws:path mqtt -t PONTOS_HUB/#
     assert_line --partial 'received CONNACK MqttConnAck{reasonCode=SUCCESS'
     assert_line --partial 'received SUBACK MqttSubAck{reasonCodes=[GRANTED_QOS_2]'
 
@@ -145,3 +145,29 @@ teardown_file() {
     assert_output --partial 'test_vessel'
 }
 
+
+@test "AUTH: mqtt editor" {
+
+    # Lets generate a token for the publisher
+    publisher_token=$(jwt encode --sub=__token__ --secret="$PONTOS_JWT_SECRET" '{"acl":{"pub": ["PONTOS/test_vessel/test_parameter/+"]}}')
+
+    # And fetch one for the subscriber
+    run curl -X POST --location --silent localhost/token
+    assert_equal "$status" 0
+    subscriber_token="$output"
+
+    # Start a subscriber in the background and let it run for 10s
+    docker run --name subscriber --detach --network='host' hivemq/mqtt-cli:4.15.0 sub -v -h localhost -p 80 -u '__token__' -pw "$subscriber_token" -ws -ws:path mqtt -t PONTOS_HUB/#
+
+    # Publish an actual payload that should be rewritten by the mqtt editor
+    run docker run --network='host' hivemq/mqtt-cli:4.15.0 pub -v -h localhost -p 80 -u '__token__' -pw "$publisher_token" -ws -ws:path mqtt -t PONTOS/test_vessel/test_parameter/1 -m '{"timestamp": 12345678, "value": 42}'
+    assert_line --partial 'received CONNACK MqttConnAck{reasonCode=SUCCESS'
+    assert_line --partial 'received PUBLISH acknowledgement'
+
+    sleep 1
+
+    run docker logs subscriber
+    assert_line --partial '{"timestamp": 12345678, "value": 42}'
+
+    docker kill subscriber
+}

tests/22-test-setup-with-custom-acl-rules.bats

@@ -28,7 +28,7 @@ teardown() {
 
     # Start base setup
     docker compose -f docker-compose.base.yml -f docker-compose.auth.yml up -d --build
-    sleep 10
+    sleep 15
 
     # Lets generate a token by ourselves with subject 'test_user'
     token=$(jwt encode --sub=test_user --secret="$PONTOS_JWT_SECRET")