[tipatterson-dev]

Accepts MQTT wildcard topic filters (+, #) on Resource Event and Batch Resource Event topics (/req/mqtt/subscriptions). A filter covers every authorized matching topic, also for resources created later, and users get only the events of resources they can read through the CS API.

Closes #255. Closes #269. Part of #249.

Stacked on #266 (Batch Resource Events, #258), which is merged. The base is master. Also depends on #262 (#257), which is merged into this branch: the delivery check below trusts the list of accepted exact subscriptions, and #262 stops a rejected subscription from being added to that list. #262, #276, #261, #263, #264 and #266 are merged.

This PR also contains the fix for #269 (subscriptions after a handler restart). The fix was in #277, which is closed: the code here is the same, and the wildcard check for restored subscriptions needs it. See "Fix for #269" below.

Module split (#274)

The Part 3 binding is in the new module sensorhub-service-csapi-mqtt (#274, from #261 up). No PR in the stack changes sensorhub-service-consys-mqtt, so nodes that use the legacy topics do not change, in any merge order.

Why a check for each message

A wildcard filter can match the topics of resources that the user cannot read. The broker sends each matching message to each matching subscriber, so a check at subscribe time is not sufficient (/req/mqtt/subscriptions: users SHALL NOT receive resource information through MQTT that they are not authorized to retrieve through HTTP). The server now checks each message before it is sent through a wildcard subscription.

Changes

• sensorhub-comm-mqtt 2.1.0: IMqttHandler.supportsWildcards() and IMqttHandler.canDeliver(userID, topic). Both default to false, so other handlers (for example SensorThings) keep rejecting wildcards.
• sensorhub-service-mqtt-hivemq 0.2.0:
  • New OshDeliveryFilter (HiveMQ outbound publish interceptor). A message on a topic that the client subscribed to exactly is sent at once. Any other message matched a wildcard filter, and the handler of its topic decides if it is sent. The check runs outside the HiveMQ threads. On error or timeout the message is not sent.
  • Wildcard filters are rejected for handlers that do not support them.
  • Fix for mqtt-hivemq: subscriptions stop getting messages after a handler module restarts #269 (from mqtt-hivemq: restore client subscriptions when a handler registers again #277): after a handler module restarted, clients that were already subscribed got no more messages, because the new handler never got onSubscribe for them. OshExtension now keeps the user of each client subscription, and registerHandler calls onSubscribe for the existing subscriptions under the prefix of the new handler. If onSubscribe fails (for example, the resource is gone or the user lost the permission), the subscription is removed from the client in HiveMQ.
  • A restored wildcard filter is removed if the new handler does not support wildcards. This is the same check as for a new SUBSCRIBE.
  • The bundled config.xml and config-tls.xml allow wildcard subscriptions. The MQTT Server does not change a config that is in the config folder. A node from an earlier version keeps its config, which does not allow wildcards, and the MQTT Server logs a warning at start.
  • README: wildcard subscriptions, config upgrade, delivery check.
• sensorhub-service-csapi-mqtt 0.2.0 (the Part 3 module, see csapi-mqtt: move the Part 3 binding to its own module #274):
  • TopicMapper.parseFilter: checks the MQTT wildcard rules, and accepts a filter only if it can match a Resource Event or Batch Resource Event topic. Filters with :data are rejected (/req/mqtt/subscriptions: the service description declares restrictions on wildcards for Resource Data topics).
  • Subscribe: the user must be able to read the resource before the first wildcard, or to list systems if the filter names no resource.
  • canDeliver: Resource Data is never sent through a wildcard subscription. An event is sent only if the user has the permission to read its resource, directly or from a system above it. This is the permission check of a CS API GET by ID (hasPermission), without a read of the resource. For a deleted datastream or control stream, osh-core cannot find the systems above it, so the systems of the topic path are used. Decisions are kept for 10 seconds in a Guava cache (10,000 entries at most).
  • README and AsyncAPI description: wildcard support, the Resource Data restriction, access rules and known limits.

Known limits

• Only the permission is checked. If the CS API hides a resource (exposedResources), a user with the permission still gets its events.
• Access decisions for wildcard subscriptions are kept for 10 seconds. Exact subscriptions are checked at subscribe time only, as before.
• {nodeId}/systems/{id} carries top-level systems only. A subscription with the ID of a subsystem gets no events.
• A filter that ends with # also matches the :data topics that other clients subscribe to. These messages are checked and not sent, which adds load with high-rate data.

Testing

• TestOshExtension: exact subscriptions bypass the check, wildcard matches ask the handler, no handler or a handler error drops the message, handlers without wildcard support reject filters. For mqtt-hivemq: subscriptions stop getting messages after a handler module restarts #269: a restarted handler gets the existing subscriptions with the correct users, a new SUBSCRIBE to the same topic does not count the client twice, a subscription that the handler refuses is removed from the client, only the topics of the new handler are restored, and a restored wildcard filter is removed when the new handler has no wildcard support.
• TestMqttServerConfig: the bundled config allows wildcards, a config that disables them is flagged (also with other line endings and indentation), a config without the element is not flagged.
• TestTopicMapper: accepted filters and the resource to check, invalid filters (syntax, :data, no matching event topic, unsupported types).
• TestWildcardDelivery: readable and unreadable resources, permission on the resource or on a system above it, deleted resources (systems of the topic path), batch events checked on the datastream, no Resource Data, cache expiry and size limit.
• TestAsyncApiHandler: wildcard section in the description.
• Each commit builds and passes the tests of the three modules on its own. SensorThings compiles.
• On a test node with CS API access control on, and a user denied two systems (run before the review changes: the delivery check then used a GET, and has not run on a node since):
  • anonymous node1/systems/+: not authorized; filter under a denied system: not authorized; :data filter and a filter that matches no event topic: rejected
  • admin node1/systems/+: create and delete events of both new systems
  • limited user node1/systems/+: create and delete of the allowed system only; nothing for the denied system
  • limited user node1/#: batch events of readable datastreams and events of the allowed system; no :data messages while another client had an exact :data subscription
  • node with a changed HiveMQ config that disables wildcards: warning at start, wildcard filters rejected by the broker
  • the AsyncAPI description parses as YAML
• Module split (csapi-mqtt: move the Part 3 binding to its own module #274), on a test node with both modules: legacy topics under /api/ and Part 3 data, events and wildcards under api/ work, with no cross-talk; with only one module, the other module's topics are rejected; an old config that names the legacy classes starts the legacy module

Stack

1. mqtt-hivemq: re-check permissions when a rejected subscribe is retried #262 Retried subscribe re-checks permissions (mqtt-hivemq: a retried subscribe skips the permission check #257) (merged)
2. comm-mqtt: clear the "Waiting for MQTT server..." status when the server is already started #276 Fix for comm-mqtt: MQTT service shows "Waiting for MQTT server..." after it has started #268, MQTT service status after start (merged)
3. mqtt-hivemq: restore client subscriptions when a handler registers again #277 Fix for mqtt-hivemq: subscriptions stop getting messages after a handler module restarts #269, subscriptions after a handler restart (closed, the fix is in this PR)
4. csapi-mqtt: Part 3 topic scheme (new module, 0.1.0) #261 Part 3 topic scheme (merged)
5. csapi-mqtt: AsyncAPI service description #263 AsyncAPI service description (merged)
6. csapi-mqtt: Resource Events for systems, datastreams and control streams #264 Resource Events (merged)
7. csapi-mqtt: Batch Resource Events for observations #266 Batch Resource Events (merged)
8. This PR

[alexrobin]

Why do we need to check old default configs SHAs? Can we just turn wildcards off if they are disabled in config or below xml is completely missing from config file?

[alexrobin]

Oh I see why now. I don't think we should enable wildcards automatically if an older default config is used. It can be confusing. Did you have a use case in mind?

[tipatterson-dev]

No, honestly, this was a missed Claude-ism

[alexrobin]

We should use a Guava cache instead of reinventing the wheel here. It would greatly simplify the code. Guava caches can be configure to expire entries after a certain time of when a certain number of entries is reached.

[tipatterson-dev]

I am not familiar with those or really Guava in general. I will look into it and switch over once I do some learning.

[alexrobin]

We could avoid throwing exceptions here by doing our own check with security.hasPermission() like in doCheckDeleted()

[alexrobin]

And if we do that, I don't think we need a separate a doCheckDeleted method actually.

[tipatterson-dev]

Rebased on master and addressed the review:

• 5fb64621 Guava cache and hasPermission (csapi-mqtt): the access cache is now a Guava cache (10 s, 10,000 entries). The delivery check calls security.hasPermission() in place of a GET, and doCheckDeleted is removed.
• 129ddfbe No automatic config replace (mqtt-hivemq): the SHA check is removed. An existing HiveMQ config is never changed. If it disables wildcards, the MQTT Server logs a warning at start.
• 3edd9222 Version (csapi-mqtt): 0.2.0.