Initial: Node-RED flow validator with conftest/Rego
- Declarative YAML flow format - Python flow2json converter with auto-positioning - Rego policies: 14 validation rules for flow integrity - Conftest integration - justfile workflow: generate → validate → deploy
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
policies:
|
||||
- policy/
|
||||
@@ -0,0 +1,48 @@
|
||||
version: 1
|
||||
|
||||
brokers:
|
||||
mosquitto:
|
||||
host: mosquitto
|
||||
port: 1883
|
||||
|
||||
dashboards:
|
||||
- tab: Home
|
||||
icon: dashboard
|
||||
groups:
|
||||
- name: Zigbee
|
||||
width: 6
|
||||
|
||||
flows:
|
||||
- tab: Zigbee Monitor
|
||||
description: Monitor and control zigbee2mqtt coordinator
|
||||
nodes:
|
||||
- id: zigbee-in
|
||||
type: mqtt-in
|
||||
broker: mosquitto
|
||||
topic: zigbee2mqtt/#
|
||||
qos: 2
|
||||
|
||||
- id: debug
|
||||
type: debug
|
||||
|
||||
- id: last-msg
|
||||
type: ui-text
|
||||
group: Zigbee
|
||||
format: "{{msg.payload|json}}"
|
||||
|
||||
- id: permit-btn
|
||||
type: ui-button
|
||||
group: Zigbee
|
||||
label: "Permit Join (60s)"
|
||||
topic: zigbee2mqtt/bridge/request/permit_join
|
||||
payload:
|
||||
value: true
|
||||
time: 60
|
||||
|
||||
- id: mqtt-out
|
||||
type: mqtt-out
|
||||
broker: mosquitto
|
||||
|
||||
wires:
|
||||
zigbee-in: [debug, last-msg]
|
||||
permit-btn: [mqtt-out]
|
||||
@@ -0,0 +1,81 @@
|
||||
# nr-flow-validator — Declarative Node-RED flows with conftest validation
|
||||
|
||||
set export
|
||||
|
||||
# Default: generate a flow, validate, and show results
|
||||
default: validate
|
||||
|
||||
# ---- Flow generation ----
|
||||
|
||||
# Generate Node-RED flow JSON from a YAML spec
|
||||
generate FLOW="zigbee-monitor":
|
||||
@echo "=== Generating {{FLOW}} ==="
|
||||
python3 scripts/flow2json.py flows/{{FLOW}}.yaml > output/{{FLOW}}.json
|
||||
@echo "Wrote output/{{FLOW}}.json"
|
||||
|
||||
# Generate all flows
|
||||
generate-all:
|
||||
@for f in flows/*.yaml; do \
|
||||
name=$(basename "$f" .yaml); \
|
||||
just generate "$name"; \
|
||||
done
|
||||
|
||||
# ---- Validation ----
|
||||
|
||||
# Validate a generated flow against Rego policies
|
||||
validate FLOW="zigbee-monitor": generate
|
||||
@echo "=== Validating {{FLOW}} ==="
|
||||
conftest test output/{{FLOW}}.json -p policy/ --namespace nodered --no-color
|
||||
|
||||
# Validate all flows
|
||||
validate-all:
|
||||
@for f in output/*.json; do \
|
||||
echo "=== Validating $(basename "$f") ==="; \
|
||||
conftest test "$$f" -p policy/ --namespace nodered --no-color; \
|
||||
done
|
||||
|
||||
# ---- Deployment ----
|
||||
|
||||
# Deploy flow to Node-RED instance
|
||||
deploy FLOW="zigbee-monitor" HOST="http://192.168.9.147:1880": generate validate
|
||||
@echo "=== Deploying {{FLOW}} to {{HOST}} ==="
|
||||
curl -s -o /dev/null -w "HTTP %{http_code}" \
|
||||
-X POST {{HOST}}/flows \
|
||||
-H "Content-Type: application/json" \
|
||||
-H "Node-RED-Deployment-Type: full" \
|
||||
-d @output/{{FLOW}}.json
|
||||
@echo ""
|
||||
|
||||
# Quick deploy (skip validation)
|
||||
deploy-quick FLOW="zigbee-monitor" HOST="http://192.168.9.147:1880": generate
|
||||
@echo "=== Quick deploy {{FLOW}} to {{HOST}} ==="
|
||||
curl -s -o /dev/null -w "HTTP %{http_code}" \
|
||||
-X POST {{HOST}}/flows \
|
||||
-H "Content-Type: application/json" \
|
||||
-H "Node-RED-Deployment-Type: full" \
|
||||
-d @output/{{FLOW}}.json
|
||||
@echo ""
|
||||
|
||||
# ---- Viewing ----
|
||||
|
||||
# Fetch and display current flows from Node-RED
|
||||
pull HOST="http://192.168.9.147:1880":
|
||||
curl -s {{HOST}}/flows | python3 -m json.tool
|
||||
|
||||
# Open dashboard in browser
|
||||
dashboard:
|
||||
xdg-open http://192.168.9.147:1880/ui 2>/dev/null || open http://192.168.9.147:1880/ui 2>/dev/null || echo "Open: http://192.168.9.147:1880/ui"
|
||||
|
||||
# ---- Development ----
|
||||
|
||||
# Watch for changes and re-validate
|
||||
watch FLOW="zigbee-monitor":
|
||||
@while true; do \
|
||||
inotifywait -q -e modify flows/{{FLOW}}.yaml policy/*.rego; \
|
||||
clear; just validate {{FLOW}}; \
|
||||
done
|
||||
|
||||
# Create a new flow from template
|
||||
new NAME:
|
||||
@cp flows/zigbee-monitor.yaml flows/{{NAME}}.yaml
|
||||
@echo "Created flows/{{NAME}}.yaml — edit and run 'just validate {{NAME}}'"
|
||||
@@ -0,0 +1,132 @@
|
||||
[
|
||||
{
|
||||
"id": "br1f38755c",
|
||||
"z": "",
|
||||
"type": "mqtt-broker",
|
||||
"name": "mosquitto",
|
||||
"broker": "mosquitto",
|
||||
"port": "1883",
|
||||
"clientid": "",
|
||||
"autoConnect": true,
|
||||
"usetls": false,
|
||||
"protocolVersion": "4",
|
||||
"keepalive": "60",
|
||||
"cleansession": true,
|
||||
"birthTopic": "",
|
||||
"birthQos": "0",
|
||||
"birthPayload": "",
|
||||
"closeTopic": "",
|
||||
"closeQos": "0",
|
||||
"closePayload": "",
|
||||
"willTopic": "",
|
||||
"willQos": "0",
|
||||
"willPayload": ""
|
||||
},
|
||||
{
|
||||
"id": "t66ef1944",
|
||||
"z": "",
|
||||
"type": "ui_tab",
|
||||
"name": "Home",
|
||||
"icon": "dashboard",
|
||||
"order": 1,
|
||||
"disabled": false,
|
||||
"hidden": false
|
||||
},
|
||||
{
|
||||
"id": "gc40a46a5",
|
||||
"z": "t66ef1944",
|
||||
"type": "ui_group",
|
||||
"name": "Zigbee",
|
||||
"tab": "t66ef1944",
|
||||
"order": 1,
|
||||
"disp": true,
|
||||
"width": "6"
|
||||
},
|
||||
{
|
||||
"id": "zigbee-in",
|
||||
"z": "f44444498",
|
||||
"type": "mqtt-in",
|
||||
"name": "",
|
||||
"nl": false,
|
||||
"rap": true,
|
||||
"rh": 0,
|
||||
"inputs": 0,
|
||||
"datatype": "auto",
|
||||
"x": 170,
|
||||
"y": 80,
|
||||
"broker": "br1f38755c",
|
||||
"topic": "zigbee2mqtt/#",
|
||||
"qos": "2",
|
||||
"wires": [
|
||||
[
|
||||
"debug",
|
||||
"last-msg"
|
||||
]
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "debug",
|
||||
"z": "f44444498",
|
||||
"type": "debug",
|
||||
"name": "Debug",
|
||||
"active": true,
|
||||
"tosidebar": true,
|
||||
"console": false,
|
||||
"tostatus": false,
|
||||
"complete": "payload",
|
||||
"targetType": "msg",
|
||||
"statusVal": "",
|
||||
"statusType": "auto",
|
||||
"x": 770,
|
||||
"y": 80
|
||||
},
|
||||
{
|
||||
"id": "last-msg",
|
||||
"z": "f44444498",
|
||||
"type": "ui-text",
|
||||
"name": "Text",
|
||||
"width": "6",
|
||||
"height": "2",
|
||||
"layout": "row-spread",
|
||||
"x": 770,
|
||||
"y": 80,
|
||||
"group": "gc40a46a5",
|
||||
"format": "{{msg.payload|json}}"
|
||||
},
|
||||
{
|
||||
"id": "permit-btn",
|
||||
"z": "f44444498",
|
||||
"type": "ui-button",
|
||||
"name": "Permit Join (60s)",
|
||||
"width": "3",
|
||||
"height": "1",
|
||||
"payloadType": "json",
|
||||
"x": 170,
|
||||
"y": 220,
|
||||
"group": "gc40a46a5",
|
||||
"payload": "{\"value\": true, \"time\": 60}",
|
||||
"topic": "zigbee2mqtt/bridge/request/permit_join",
|
||||
"wires": [
|
||||
[
|
||||
"mqtt-out"
|
||||
]
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "mqtt-out",
|
||||
"z": "f44444498",
|
||||
"type": "mqtt-out",
|
||||
"name": "",
|
||||
"qos": "2",
|
||||
"retain": false,
|
||||
"respTopic": "",
|
||||
"contentType": "",
|
||||
"userProps": "",
|
||||
"correl": "",
|
||||
"expiry": "",
|
||||
"x": 770,
|
||||
"y": 220,
|
||||
"broker": "br1f38755c",
|
||||
"topic": ""
|
||||
}
|
||||
]
|
||||
@@ -0,0 +1,169 @@
|
||||
package nodered
|
||||
import future.keywords.if
|
||||
import future.keywords.contains
|
||||
import future.keywords.in
|
||||
|
||||
# helpers
|
||||
config_node(typ) if typ in {"mqtt-broker", "ui_tab", "ui_base"}
|
||||
welcome_isolated(typ) if typ in {"debug", "mqtt-broker", "ui_tab", "ui_group", "ui_text", "ui_chart", "ui_gauge", "ui_button", "ui_slider", "ui_switch", "inject", "comment", "mqtt in", "catch", "status", "link in"}
|
||||
|
||||
broker_node(id) if {
|
||||
some node in input
|
||||
node.type == "mqtt-broker"
|
||||
node.id == id
|
||||
}
|
||||
|
||||
ui_group(id) if {
|
||||
some node in input
|
||||
node.type == "ui_group"
|
||||
node.id == id
|
||||
}
|
||||
|
||||
ui_tab(id) if {
|
||||
some node in input
|
||||
node.type == "ui_tab"
|
||||
node.id == id
|
||||
}
|
||||
|
||||
node_exists(id) if {
|
||||
some node in input
|
||||
node.id == id
|
||||
}
|
||||
|
||||
has_inbound_wire(target_id) if {
|
||||
some node in input
|
||||
some wire in node.wires
|
||||
target_id in wire
|
||||
}
|
||||
|
||||
has_outbound_wire(source_id) if {
|
||||
some node in input
|
||||
node.id == source_id
|
||||
node.wires
|
||||
count(node.wires) > 0
|
||||
}
|
||||
|
||||
# ---- DENY RULES ----
|
||||
|
||||
deny_missing_type contains msg if {
|
||||
some node in input
|
||||
not node.type
|
||||
msg := sprintf("node %s: missing type field", [node.id])
|
||||
}
|
||||
|
||||
deny_missing_field contains msg if {
|
||||
some node in input
|
||||
node.type == "mqtt-broker"
|
||||
required := {"type", "broker", "port"}
|
||||
some f in required
|
||||
not node[f]
|
||||
msg := sprintf("node %s (mqtt-broker): missing %s", [node.id, f])
|
||||
}
|
||||
|
||||
deny_missing_field contains msg if {
|
||||
some node in input
|
||||
node.type == "mqtt in"
|
||||
required := {"type", "topic", "broker", "z"}
|
||||
some f in required
|
||||
not node[f]
|
||||
msg := sprintf("node %s (mqtt in): missing %s", [node.id, f])
|
||||
}
|
||||
|
||||
deny_missing_field contains msg if {
|
||||
some node in input
|
||||
node.type == "mqtt out"
|
||||
required := {"type", "broker", "z"}
|
||||
some f in required
|
||||
not node[f]
|
||||
msg := sprintf("node %s (mqtt out): missing %s", [node.id, f])
|
||||
}
|
||||
|
||||
deny_missing_field contains msg if {
|
||||
some node in input
|
||||
node.type == "debug"
|
||||
required := {"type", "z"}
|
||||
some f in required
|
||||
not node[f]
|
||||
msg := sprintf("node %s (debug): missing %s", [node.id, f])
|
||||
}
|
||||
|
||||
deny_missing_field contains msg if {
|
||||
some node in input
|
||||
node.type in {"ui_text", "ui_button", "ui_chart", "ui_gauge", "ui_slider", "ui_switch"}
|
||||
required := {"type", "group", "z"}
|
||||
some f in required
|
||||
not node[f]
|
||||
msg := sprintf("node %s (%s): missing %s", [node.id, node.type, f])
|
||||
}
|
||||
|
||||
deny_missing_field contains msg if {
|
||||
some node in input
|
||||
node.type == "ui_group"
|
||||
required := {"type", "tab", "name"}
|
||||
some f in required
|
||||
not node[f]
|
||||
msg := sprintf("node %s (ui_group): missing %s", [node.id, f])
|
||||
}
|
||||
|
||||
deny_missing_field contains msg if {
|
||||
some node in input
|
||||
node.type == "ui_tab"
|
||||
required := {"type", "name"}
|
||||
some f in required
|
||||
not node[f]
|
||||
msg := sprintf("node %s (ui_tab): missing %s", [node.id, f])
|
||||
}
|
||||
|
||||
deny_broker_ref contains msg if {
|
||||
some node in input
|
||||
node.type in {"mqtt in", "mqtt out"}
|
||||
not broker_node(node.broker)
|
||||
msg := sprintf("node %s (%s): references unknown broker %s", [node.id, node.type, node.broker])
|
||||
}
|
||||
|
||||
deny_group_ref contains msg if {
|
||||
some node in input
|
||||
node.type in {"ui_text", "ui_button", "ui_chart", "ui_gauge", "ui_slider", "ui_switch"}
|
||||
not ui_group(node.group)
|
||||
msg := sprintf("node %s (%s): references unknown group %s", [node.id, node.type, node.group])
|
||||
}
|
||||
|
||||
deny_tab_ref contains msg if {
|
||||
some node in input
|
||||
node.type == "ui_group"
|
||||
not ui_tab(node.tab)
|
||||
msg := sprintf("node %s (ui_group): references unknown tab %s", [node.id, node.tab])
|
||||
}
|
||||
|
||||
deny_wire_ref contains msg if {
|
||||
some node in input
|
||||
some wire in node.wires
|
||||
some target in wire
|
||||
not node_exists(target)
|
||||
msg := sprintf("node %s (%s): wire to unknown node %s", [node.id, node.type, target])
|
||||
}
|
||||
|
||||
deny_missing_flow contains msg if {
|
||||
some node in input
|
||||
not config_node(node.type)
|
||||
not node.z
|
||||
msg := sprintf("node %s (%s): missing flow assignment (z)", [node.id, node.type])
|
||||
}
|
||||
|
||||
deny_isolated_node contains msg if {
|
||||
some node in input
|
||||
not welcome_isolated(node.type)
|
||||
not has_inbound_wire(node.id)
|
||||
not has_outbound_wire(node.id)
|
||||
msg := sprintf("node %s (%s): has no wires", [node.id, node.type])
|
||||
}
|
||||
|
||||
# MQTT brokers must have valid port
|
||||
deny_broker_port contains msg if {
|
||||
some node in input
|
||||
node.type == "mqtt-broker"
|
||||
not is_valid_port(to_number(node.port))
|
||||
msg := sprintf("node %s (mqtt-broker): invalid port %s", [node.id, node.port])
|
||||
}
|
||||
|
||||
is_valid_port(p) if { p >= 1; p <= 65535 }
|
||||
@@ -0,0 +1,164 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Convert declarative flow YAML to Node-RED flow JSON."""
|
||||
import json
|
||||
import sys
|
||||
import uuid
|
||||
import yaml
|
||||
|
||||
COLUMNS = [170, 450, 770]
|
||||
ROW_HEIGHT = 140
|
||||
BROKER_DEFAULTS = {
|
||||
"clientid": "", "autoConnect": True, "usetls": False,
|
||||
"protocolVersion": "4", "keepalive": "60", "cleansession": True,
|
||||
"birthTopic": "", "birthQos": "0", "birthPayload": "",
|
||||
"closeTopic": "", "closeQos": "0", "closePayload": "",
|
||||
"willTopic": "", "willQos": "0", "willPayload": "",
|
||||
}
|
||||
|
||||
NODE_TEMPLATES = {
|
||||
"mqtt-in": {
|
||||
"nl": False, "rap": True, "rh": 0, "inputs": 0, "datatype": "auto",
|
||||
},
|
||||
"mqtt-out": {
|
||||
"qos": "2", "retain": False, "respTopic": "", "contentType": "",
|
||||
"userProps": "", "correl": "", "expiry": "",
|
||||
},
|
||||
"debug": {
|
||||
"active": True, "tosidebar": True, "console": False,
|
||||
"tostatus": False, "complete": "payload", "targetType": "msg",
|
||||
"statusVal": "", "statusType": "auto",
|
||||
},
|
||||
"ui-text": {
|
||||
"width": "6", "height": "2", "layout": "row-spread",
|
||||
},
|
||||
"ui-button": {
|
||||
"width": "3", "height": "1", "payloadType": "json",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def new_id(prefix="n"):
|
||||
return prefix + uuid.uuid4().hex[:8]
|
||||
|
||||
|
||||
def make_broker_node(name, cfg, flow_id):
|
||||
node = {"id": new_id("br"), "z": "", "type": "mqtt-broker", "name": name}
|
||||
node["broker"] = cfg["host"]
|
||||
node["port"] = str(cfg.get("port", 1883))
|
||||
for k, v in BROKER_DEFAULTS.items():
|
||||
node.setdefault(k, v)
|
||||
return node
|
||||
|
||||
|
||||
def make_node(nd, col, row, broker_map, group_map, flow_id):
|
||||
ntype = nd["type"]
|
||||
nid = nd["id"] if "id" in nd else new_id()
|
||||
node = {"id": nid, "z": flow_id, "type": ntype, "name": nd.get("label", nd.get("name", ""))}
|
||||
node.update(NODE_TEMPLATES.get(ntype, {}))
|
||||
node["x"] = COLUMNS[col] if col < len(COLUMNS) else col * 300 + 170
|
||||
node["y"] = row * ROW_HEIGHT + 80
|
||||
|
||||
if ntype in ("mqtt-in", "mqtt-out"):
|
||||
node["broker"] = broker_map[nd["broker"]]
|
||||
if ntype in ("ui-text", "ui-button"):
|
||||
node["group"] = group_map[nd["group"]]
|
||||
if ntype == "mqtt-in":
|
||||
node["topic"] = nd["topic"]
|
||||
node["qos"] = str(nd.get("qos", 2))
|
||||
if ntype == "mqtt-out":
|
||||
node["topic"] = nd.get("topic", "")
|
||||
node["qos"] = str(nd.get("qos", 2))
|
||||
if ntype == "ui-text":
|
||||
node["format"] = nd.get("format", "{{msg.payload}}")
|
||||
node["name"] = nd.get("label", nd.get("name", "Text"))
|
||||
if ntype == "ui-button":
|
||||
node["payload"] = json.dumps(nd.get("payload", ""))
|
||||
node["topic"] = nd.get("topic", "")
|
||||
node["name"] = nd.get("label", nd.get("name", "Button"))
|
||||
if ntype == "debug":
|
||||
node["name"] = nd.get("label", nd.get("name", "Debug"))
|
||||
|
||||
return node
|
||||
|
||||
|
||||
def make_group_node(name, tab_id, order):
|
||||
return {
|
||||
"id": new_id("g"), "z": tab_id, "type": "ui_group",
|
||||
"name": name, "tab": tab_id, "order": order,
|
||||
"disp": True, "width": "6",
|
||||
}
|
||||
|
||||
|
||||
def make_tab_node(name, icon, order):
|
||||
tab_id = new_id("t")
|
||||
return tab_id, {
|
||||
"id": tab_id, "z": "", "type": "ui_tab",
|
||||
"name": name, "icon": icon, "order": order,
|
||||
"disabled": False, "hidden": False,
|
||||
}
|
||||
|
||||
|
||||
def build_flow(doc, flow_def, node_positions):
|
||||
nodes = []
|
||||
flow_id = new_id("f")
|
||||
|
||||
# Collect broker references
|
||||
broker_map = {}
|
||||
for nd in flow_def["nodes"]:
|
||||
if nd["type"] in ("mqtt-in", "mqtt-out") and "broker" in nd:
|
||||
bname = nd["broker"]
|
||||
if bname not in broker_map:
|
||||
bcfg = doc["brokers"][bname]
|
||||
bn = make_broker_node(bname, bcfg, flow_id)
|
||||
broker_map[bname] = bn["id"]
|
||||
nodes.append(bn)
|
||||
|
||||
# Collect dashboard groups/tabs
|
||||
group_map = {}
|
||||
if "dashboards" in doc:
|
||||
for db in doc["dashboards"]:
|
||||
tab_id, tn = make_tab_node(db["tab"], db.get("icon", "dashboard"), db.get("order", 1))
|
||||
nodes.append(tn)
|
||||
for gi, grp in enumerate(db.get("groups", [])):
|
||||
gn = make_group_node(grp["name"], tab_id, gi + 1)
|
||||
group_map[grp["name"]] = gn["id"]
|
||||
nodes.append(gn)
|
||||
|
||||
# Create flow nodes with auto-positioning
|
||||
for i, nd in enumerate(flow_def["nodes"]):
|
||||
col, row = node_positions.get(nd["id"], (0, i))
|
||||
node = make_node(nd, col, row, broker_map, group_map, flow_id)
|
||||
nodes.append(node)
|
||||
|
||||
# Wire connections
|
||||
wires = flow_def.get("wires", {})
|
||||
for src_id, targets in wires.items():
|
||||
src = next(n for n in nodes if n.get("id") == src_id or n.get("name", "").lower().replace(" ", "-") == src_id)
|
||||
src.setdefault("wires", [])
|
||||
src["wires"].append(targets)
|
||||
|
||||
return nodes
|
||||
|
||||
|
||||
def auto_position(flow_def):
|
||||
"""Simple auto-layout: put each node in a column determined by its role."""
|
||||
positions = {}
|
||||
col_order = {"mqtt-in": 0, "ui-button": 0, "inject": 0,
|
||||
"debug": 2, "ui-text": 2, "mqtt-out": 2, "ui-chart": 2, "ui-gauge": 2}
|
||||
for i, nd in enumerate(flow_def["nodes"]):
|
||||
col = col_order.get(nd["type"], 1)
|
||||
positions[nd["id"]] = (col, i // 3)
|
||||
return positions
|
||||
|
||||
|
||||
def main():
|
||||
doc = yaml.safe_load(open(sys.argv[1]))
|
||||
all_nodes = []
|
||||
for flow_def in doc.get("flows", []):
|
||||
positions = auto_position(flow_def)
|
||||
all_nodes += build_flow(doc, flow_def, positions)
|
||||
json.dump(all_nodes, sys.stdout, indent=2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user