Feat/enterprise mqtt reader block

inference/core/env.py

@@ -157,6 +157,35 @@
         for entry in KAFKA_WORKFLOWS_SINKS_WHITELISTED_BOOTSTRAP_SERVERS.split(",")
         if entry.strip()
     ]
+# Operator policy for the MQTT Reader / MQTT Writer Workflow blocks, which open
+# an outbound connection to the `host`/`port` taken from the workflow definition
+# or its inputs. When True (default, preserves behaviour) that value is
+# honoured, subject to the allowlist below. When False the workflow-provided
+# value is ignored and the blocks connect to the first entry of
+# MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS instead; with no entry configured
+# there, the MQTT blocks are disabled (every run reports an error).
+MQTT_WORKFLOWS_BLOCKS_ALLOW_USER_PROVIDED_HOST = str2bool(
+    os.getenv("MQTT_WORKFLOWS_BLOCKS_ALLOW_USER_PROVIDED_HOST", True)
+)
+# Optional comma-separated list of `host[:port]` MQTT brokers the MQTT Workflow
+# blocks may connect to. When set and a user-provided host is allowed, the
+# workflow's `host:port` must match an entry or the run reports an error; an
+# entry without a port matches that host on any port. When a user-provided host
+# is not allowed, the first entry is what the blocks connect to (port 1883 when
+# the entry has none). Hosts are compared after trimming whitespace, lowercasing
+# and stripping IPv6 brackets; there is no DNS resolution. Empty entries are
+# ignored. A variable that is set but holds no entries is an empty allowlist:
+# nothing is permitted. Default None (no allowlist).
+MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS = os.getenv(
+    "MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS"
+)
+if MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS is not None:
+    # a list, not a set: the first entry is the operator's broker
+    MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS = [
+        entry.strip()
+        for entry in MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS.split(",")
+        if entry.strip()
+    ]
 
 # List of allowed origins
 ALLOW_ORIGINS = os.getenv("ALLOW_ORIGINS", "*")

inference/core/interfaces/workflows_configuration.py

@@ -79,6 +79,14 @@ def build_configuration_from_env() -> WorkflowsConfiguration:
                 if env.KAFKA_WORKFLOWS_SINKS_WHITELISTED_BOOTSTRAP_SERVERS is None
                 else tuple(env.KAFKA_WORKFLOWS_SINKS_WHITELISTED_BOOTSTRAP_SERVERS)
             ),
+            allow_mqtt_blocks_user_provided_host=env.MQTT_WORKFLOWS_BLOCKS_ALLOW_USER_PROVIDED_HOST,
+            # NOT sorted: the first entry is the operator's broker when
+            # user-provided hosts are not allowed.
+            mqtt_blocks_whitelisted_hosts=(
+                None
+                if env.MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS is None
+                else tuple(env.MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS)
+            ),
         ),
         tensor=TensorConfiguration(
             representation_enabled=env.ENABLE_TENSOR_DATA_REPRESENTATION,

tests/inference/unit_tests/core/interfaces/test_workflows_configuration.py

@@ -99,6 +99,18 @@
             else list(c.engine.kafka_sinks_whitelisted_bootstrap_servers)
         ),
     ),
+    (
+        "MQTT_WORKFLOWS_BLOCKS_ALLOW_USER_PROVIDED_HOST",
+        lambda c: c.engine.allow_mqtt_blocks_user_provided_host,
+    ),
+    (
+        "MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS",
+        lambda c: (
+            None
+            if c.engine.mqtt_blocks_whitelisted_hosts is None
+            else list(c.engine.mqtt_blocks_whitelisted_hosts)
+        ),
+    ),
     ("ENABLE_TENSOR_DATA_REPRESENTATION", lambda c: c.tensor.representation_enabled),
     ("WORKFLOWS_IMAGE_TENSOR_DEVICE", lambda c: c.tensor.image_tensor_device),
     (
@@ -226,7 +238,7 @@ def test_the_field_table_matches_the_facade_exports() -> None:
         "missing_from_table": sorted(exported - tabled),
         "missing_from_facade": sorted(tabled - exported),
     }
-    assert len(tabled) == 78, len(tabled)
+    assert len(tabled) == 80, len(tabled)
 
 
 def test_every_name_workflows_imports_from_the_facade_is_exported() -> None:
@@ -291,6 +303,11 @@ def test_server_configuration_equals_env_field_by_field(name, reader) -> None:
             "KAFKA_WORKFLOWS_SINKS_WHITELISTED_BOOTSTRAP_SERVERS",
             ["kafka-2:9092", "kafka-1:9092"],
         ),
+        ("MQTT_WORKFLOWS_BLOCKS_ALLOW_USER_PROVIDED_HOST", False),
+        ("MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS", None),
+        ("MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS", []),
+        # order preserved, not sorted
+        ("MQTT_WORKFLOWS_BLOCKS_WHITELISTED_HOSTS", ["broker-2:1883", "broker-1"]),
         ("LAMBDA", True),
     ],
 )