diff --git a/lib/assistant.js b/lib/assistant.js index 28d1585a..c080cbe1 100644 --- a/lib/assistant.js +++ b/lib/assistant.js @@ -4,7 +4,7 @@ const { z } = require('zod') const auth = require('./auth') const { getLongestUpstreamPath } = require('./flowGraph') -const { hasProperty } = require('./utils') +const { hasProperty, summarizeMsg } = require('./utils') const semver = require('semver') // import typedef AssistantSettings @@ -465,6 +465,10 @@ class Assistant { return assistant.handleGetDeployPolicyRequest(req, res) }) + RED.httpAdmin.post('/nr-assistant/inject', RED.auth.needsPermission('write'), function (req, res) { + return assistant.handlePostInjectRequest(req, res) + }) + RED.httpAdmin.post('/nr-assistant/:method', RED.auth.needsPermission('write'), function (req, res) { return assistant.handlePostMethodRequest(req, res) }) @@ -558,6 +562,14 @@ class Assistant { if (!this.isInitialized || this.isLoading) { return res.status(503).send('Assistant is not ready') } + res.json({ status: 'ok', autoDeploy: await this.getAutoDeployPolicy() }) + } + + /** + * Ask forge whether the team has opted in to agent-initiated deploys. Fails closed. + * @returns {Promise} true if agent-initiated deploys are allowed + */ + async getAutoDeployPolicy () { const url = `${this.options.url.replace(/\/$/, '')}/deploy-policy` try { const response = await this.got(url, { @@ -568,11 +580,57 @@ class Assistant { }, responseType: 'json' }) - res.json({ status: 'ok', autoDeploy: !!response.body?.autoDeploy }) + return !!response.body?.autoDeploy } catch (error) { this.RED.log.warn(`nr-assistant: failed to check deploy policy: ${error.message}`) - res.json({ status: 'ok', autoDeploy: false }) + return false + } + } + + /** + * Handles POST requests to the /nr-assistant/inject endpoint. + * Injects a message into a deployed node, as if it had arrived on the node's input, so the + * agent can exercise a flow and read the results back from the debug sidebar. + * Gated on the same team opt-in as agent-initiated deploys, checked live on every call. + * @param {import('express').Request} req - The request object + * @param {import('express').Response} res - The response object + */ + async handlePostInjectRequest (req, res) { + if (!this.isInitialized || this.isLoading) { + return res.status(503).send('Assistant is not ready') } + const { id, msg } = req.body || {} + if (!id || typeof id !== 'string') { + return res.status(400).json({ status: 'error', message: 'id is required' }) + } + if (msg !== undefined && (msg === null || typeof msg !== 'object' || Array.isArray(msg))) { + return res.status(400).json({ status: 'error', message: 'msg must be an object' }) + } + if (!await this.getAutoDeployPolicy()) { + return res.json({ status: 'ok', injected: false, code: 'INJECT_NOT_ALLOWED' }) + } + const node = this.RED.nodes.getNode(id) + if (!node || typeof node.receive !== 'function') { + return res.status(404).json({ status: 'error', message: `Node ${id} not found in the deployed flows` }) + } + const injectMsg = { ...msg, _msgid: this.RED.util.generateId() } + if (typeof this.RED.log.audit === 'function') { + // NR_FLOW_NAME resolves to the label of the node's own flow + const flowName = this.RED.util.evaluateEnvProperty?.('NR_FLOW_NAME', node) + // `msg` is reserved for the log text by RED.log, hence `msgSummary` + this.RED.log.audit({ + event: 'nr-assistant.inject', + node: id, + type: node.type, + name: node.name, + z: node.z, + flowName, + msgid: injectMsg._msgid, + msgSummary: summarizeMsg(injectMsg) + }, req) + } + node.receive(injectMsg) + res.json({ status: 'ok', injected: true, msgid: injectMsg._msgid }) } /** diff --git a/lib/utils.js b/lib/utils.js index d47bb52e..e0c18ee1 100644 --- a/lib/utils.js +++ b/lib/utils.js @@ -14,8 +14,48 @@ const isObject = (obj) => { return obj !== null && typeof obj === 'object' && !Array.isArray(obj) } +const MSG_SUMMARY_MAX_PROPERTIES = 20 +const MSG_SUMMARY_MAX_TOPIC_LENGTH = 100 + +const valueType = (value) => { + if (value === null) return 'null' + if (Array.isArray(value)) return 'array' + if (Buffer.isBuffer(value)) return 'buffer' + return typeof value +} + +/** + * Summarise a message for an audit log without recording its values: the top-level property + * names and types, the topic (string or number only, truncated) and the JSON size in bytes. + * @param {Object} msg - the message to summarise (`_msgid` is ignored) + * @returns {{properties: Object, truncated?: boolean, topic?: string|number, size?: number}} + */ +const summarizeMsg = (msg) => { + const keys = Object.keys(msg || {}).filter(key => key !== '_msgid') + const properties = {} + for (const key of keys.slice(0, MSG_SUMMARY_MAX_PROPERTIES)) { + properties[key] = valueType(msg[key]) + } + const summary = { properties } + if (keys.length > MSG_SUMMARY_MAX_PROPERTIES) { + summary.truncated = true + } + if (typeof msg?.topic === 'string') { + summary.topic = msg.topic.slice(0, MSG_SUMMARY_MAX_TOPIC_LENGTH) + } else if (typeof msg?.topic === 'number') { + summary.topic = msg.topic + } + try { + summary.size = Buffer.byteLength(JSON.stringify(msg || {})) + } catch (_err) { + // not serialisable (e.g. circular) - leave size out + } + return summary +} + module.exports = { tryJsonParse, hasProperty, - isObject + isObject, + summarizeMsg } diff --git a/resources/expertAutomations.js b/resources/expertAutomations.js index 7dce11ae..baa9e6e7 100644 --- a/resources/expertAutomations.js +++ b/resources/expertAutomations.js @@ -42,6 +42,7 @@ const GET_DEBUG_MESSAGES = 'automation/get-debug-messages' const UNDO = 'automation/undo' const GET_HISTORY = 'automation/get-history' const REDO = 'automation/redo' +const INJECT_MESSAGE = 'automation/inject-message' const ALIGNMENT_DIRECTIONS = ['grid', 'left', 'right', 'top', 'bottom', 'middle', 'center'] const DISTRIBUTE_DIRECTIONS = ['horizontally', 'vertically'] @@ -64,7 +65,8 @@ const RESULT_CODES = Object.freeze({ NO_UNDEPLOYED_CHANGES: 'NO_UNDEPLOYED_CHANGES', AUTO_DEPLOY_DISABLED: 'AUTO_DEPLOY_DISABLED', DEPLOY_NOT_CONFIRMED: 'DEPLOY_NOT_CONFIRMED', - INSTALL_STARTED: 'INSTALL_STARTED' + INSTALL_STARTED: 'INSTALL_STARTED', + INJECT_NOT_ALLOWED: 'INJECT_NOT_ALLOWED' }) const LINK_NODE_TYPES = ['link in', 'link out', 'link call'] @@ -141,7 +143,8 @@ const DEPLOY_WAIT_TIMEOUT_MS = 10000 * |TOGGLE_SIDEBAR * |GET_DEBUG_MESSAGES * |UNDO - * |GET_HISTORY} ExpertAutomationsActionsEnum + * |GET_HISTORY + * |INJECT_MESSAGE} ExpertAutomationsActionsEnum */ export class ExpertAutomations extends ExpertActionsInterface { @@ -781,6 +784,22 @@ export class ExpertAutomations extends ExpertActionsInterface { } } } + }, + [INJECT_MESSAGE]: { + params: { + type: 'object', + properties: { + id: { + type: 'string', + description: 'The ID of the deployed node to inject the message into. The node must have an input port and no undeployed changes.' + }, + msg: { + type: 'object', + description: 'The message to inject, e.g. { "payload": "hello", "topic": "test" }. Defaults to {}. A new _msgid is always assigned.' + } + }, + required: ['id'] + } } }) @@ -1936,6 +1955,35 @@ export class ExpertAutomations extends ExpertActionsInterface { return this.RED.nodes.node(id) || this.RED.nodes.junction(id) } + /** + * Check a node can receive an injected message, using what only the editor knows (input ports, + * disabled state, undeployed changes). The runtime confirms the node is actually deployed. + * @param {string} id - the node id + * @returns {string|null} an error message, or null if the node is a valid inject target + */ + _getInjectTargetError (id) { + const node = this._resolveNode(id) + if (!node) { + return `Node ${id} not found` + } + if (this.RED.nodes.subflow?.(node.z)) { + // the runtime instantiates subflow internals under generated ids, one set per instance + return `Node ${id} is inside a subflow definition - inject into the subflow instance instead` + } + const inputs = node.inputs ?? node._def?.inputs ?? 0 + if (!inputs) { + return `Node ${id} (${node.type}) has no input port` + } + if (node.d || this.RED.nodes.workspace(node.z)?.disabled) { + return `Node ${id} is disabled so is not running` + } + if (node.changed) { + // the runtime still has the last deployed version of this node (or none at all) + return `Node ${id} has undeployed changes - deploy first, then inject` + } + return null + } + closeSearch () { this.RED.search.hide() } closeTypeSearch () { @@ -3299,6 +3347,38 @@ export class ExpertAutomations extends ExpertActionsInterface { result.data = this.redo(params) result.success = true break + case INJECT_MESSAGE: { + const targetError = this._getInjectTargetError(params.id) + if (targetError) { + result.success = false + result.error = targetError + break + } + // The runtime is the enforcement point: it re-checks the team's agent deploy opt-in on + // every call, so the editor does not need to (and cannot be relied upon to) gate this. + let response + try { + response = await $.ajax({ + url: 'nr-assistant/inject', + method: 'POST', + contentType: 'application/json', + data: JSON.stringify({ id: params.id, msg: params.msg || {} }), + headers: { Accept: 'application/json' } + }) + } catch (err) { + result.success = false + result.error = err?.responseJSON?.message || err?.message || 'Inject request failed' + break + } + result.success = true + result.injected = !!response?.injected + if (result.injected) { + result.msgid = response.msgid + } else { + result.code = response?.code || RESULT_CODES.INJECT_NOT_ALLOWED + } + break + } default: result.handled = false result.success = false diff --git a/test/unit/lib/assistant.test.js b/test/unit/lib/assistant.test.js index 7e5472b9..f195b2e3 100644 --- a/test/unit/lib/assistant.test.js +++ b/test/unit/lib/assistant.test.js @@ -244,6 +244,7 @@ describe('assistant', () => { RED.httpAdmin._postEndpoints.should.have.property('/nr-assistant/fim/:nodeModule/:nodeType') RED.httpAdmin._getEndpoints.should.have.property('/nr-assistant/mcp/prompts') RED.httpAdmin._postEndpoints.should.have.property('/nr-assistant/mcp/prompts/:promptId') + RED.httpAdmin._postEndpoints.should.have.property('/nr-assistant/inject') RED.httpAdmin._postEndpoints.should.have.property('/nr-assistant/:method') }) @@ -370,4 +371,93 @@ describe('assistant', () => { RED.comms.publish.firstCall.args[0].should.equal('nr-assistant/initialise') RED.comms.publish.secondCall.args[0].should.equal('nr-assistant/mcp/ready') }) + + describe('inject endpoint', () => { + let node + let res + beforeEach(async () => { + const options = { ...RED.settings.flowforge.assistant, got: fakeGot } + await assistant.init(RED, options) + node = { id: 'node-1', type: 'function', name: 'double payload', z: 'tab-1', receive: sinon.stub() } + RED.nodes = { getNode: sinon.stub().callsFake(id => id === 'node-1' ? node : null) } + RED.util = { + generateId: sinon.stub().returns('new-msg-id'), + evaluateEnvProperty: sinon.stub().callsFake((name, n) => name === 'NR_FLOW_NAME' && n === node ? 'API' : undefined) + } + RED.log.audit = sinon.stub() + sinon.stub(assistant, 'getAutoDeployPolicy').resolves(true) + res = { + status: sinon.stub().returnsThis(), + json: sinon.stub().returnsThis(), + send: sinon.stub().returnsThis() + } + }) + afterEach(() => { + delete RED.nodes + delete RED.util + delete RED.log.audit + }) + + it('injects the msg into the deployed node with a new _msgid', async () => { + await assistant.handlePostInjectRequest({ body: { id: 'node-1', msg: { payload: 'hello', _msgid: 'caller-id' } } }, res) + node.receive.calledOnce.should.be.true() + node.receive.firstCall.args[0].should.eql({ payload: 'hello', _msgid: 'new-msg-id' }) + res.json.calledWith({ status: 'ok', injected: true, msgid: 'new-msg-id' }).should.be.true() + RED.log.audit.calledOnce.should.be.true() + }) + it('audits the target node and a value-free summary of the msg', async () => { + const req = { body: { id: 'node-1', msg: { payload: 21, topic: 'sensors/line1' } } } + await assistant.handlePostInjectRequest(req, res) + const [entry, auditReq] = RED.log.audit.firstCall.args + entry.should.eql({ + event: 'nr-assistant.inject', + node: 'node-1', + type: 'function', + name: 'double payload', + z: 'tab-1', + flowName: 'API', + msgid: 'new-msg-id', + msgSummary: { + properties: { payload: 'number', topic: 'string' }, + topic: 'sensors/line1', + size: Buffer.byteLength(JSON.stringify({ payload: 21, topic: 'sensors/line1', _msgid: 'new-msg-id' })) + } + }) + auditReq.should.equal(req) + }) + it('audits without a flow name when the flow has no label', async () => { + RED.util.evaluateEnvProperty = sinon.stub() + await assistant.handlePostInjectRequest({ body: { id: 'node-1' } }, res) + const [entry] = RED.log.audit.firstCall.args + entry.z.should.equal('tab-1') + should.not.exist(entry.flowName) + }) + it('injects an empty msg when none is provided', async () => { + await assistant.handlePostInjectRequest({ body: { id: 'node-1' } }, res) + node.receive.firstCall.args[0].should.eql({ _msgid: 'new-msg-id' }) + }) + it('does not inject when the team has not enabled agent-initiated deploy', async () => { + assistant.getAutoDeployPolicy.resolves(false) + await assistant.handlePostInjectRequest({ body: { id: 'node-1', msg: {} } }, res) + node.receive.called.should.be.false() + res.json.calledWith({ status: 'ok', injected: false, code: 'INJECT_NOT_ALLOWED' }).should.be.true() + }) + it('returns 404 when the node is not in the deployed flows', async () => { + await assistant.handlePostInjectRequest({ body: { id: 'missing', msg: {} } }, res) + res.status.calledWith(404).should.be.true() + }) + it('returns 400 when id is missing', async () => { + await assistant.handlePostInjectRequest({ body: { msg: {} } }, res) + res.status.calledWith(400).should.be.true() + assistant.getAutoDeployPolicy.called.should.be.false() + }) + it('returns 400 when msg is not an object', async () => { + for (const msg of ['text', [1, 2], null]) { + res.status.resetHistory() + await assistant.handlePostInjectRequest({ body: { id: 'node-1', msg } }, res) + res.status.calledWith(400).should.be.true() + } + node.receive.called.should.be.false() + }) + }) }) diff --git a/test/unit/lib/utils.test.js b/test/unit/lib/utils.test.js index 2d725cae..73903be6 100644 --- a/test/unit/lib/utils.test.js +++ b/test/unit/lib/utils.test.js @@ -1,7 +1,7 @@ /// 'use strict' const should = require('should') -const { tryJsonParse, hasProperty, isObject } = require('../../../lib/utils') +const { tryJsonParse, hasProperty, isObject, summarizeMsg } = require('../../../lib/utils') describe('utils', function () { describe('tryJsonParse', function () { @@ -81,4 +81,46 @@ describe('utils', function () { hasProperty({ a: undefined }, 'a').should.be.true() }) }) + + describe('summarizeMsg', function () { + it('should record top-level property names and types but no values', function () { + const msg = { _msgid: 'abc', payload: 42, topic: 'sensors/line1', parts: { index: 0 }, list: [1], nothing: null, flag: true, buf: Buffer.from('x') } + const summary = summarizeMsg(msg) + summary.properties.should.eql({ payload: 'number', topic: 'string', parts: 'object', list: 'array', nothing: 'null', flag: 'boolean', buf: 'buffer' }) + summary.topic.should.equal('sensors/line1') + summary.size.should.equal(Buffer.byteLength(JSON.stringify(msg))) + should(summary.truncated).be.undefined() + JSON.stringify(summary).should.not.containEql('index') + }) + + it('should include a numeric topic', function () { + summarizeMsg({ topic: 7 }).topic.should.equal(7) + }) + + it('should leave out a topic that is not a string or number', function () { + summarizeMsg({ topic: { secret: 'x' } }).should.not.have.property('topic') + }) + + it('should truncate a long topic to 100 characters', function () { + summarizeMsg({ topic: 't'.repeat(250) }).topic.should.have.length(100) + }) + + it('should cap the number of properties and flag the truncation', function () { + const msg = Object.fromEntries(Array.from({ length: 25 }, (_, i) => [`p${i}`, i])) + const summary = summarizeMsg(msg) + Object.keys(summary.properties).should.have.length(20) + summary.truncated.should.be.true() + }) + + it('should handle an empty or missing msg', function () { + summarizeMsg({}).should.eql({ properties: {}, size: 2 }) + summarizeMsg(undefined).properties.should.eql({}) + }) + + it('should leave out size when the msg cannot be serialised', function () { + const msg = { payload: 1 } + msg.self = msg + summarizeMsg(msg).should.not.have.property('size') + }) + }) }) diff --git a/test/unit/resources/expertAutomations.test.js b/test/unit/resources/expertAutomations.test.js index 1ed56b6d..3ac48881 100644 --- a/test/unit/resources/expertAutomations.test.js +++ b/test/unit/resources/expertAutomations.test.js @@ -130,7 +130,8 @@ describeMain('expertAutomations', () => { 'automation/get-debug-messages', 'automation/undo', 'automation/get-history', - 'automation/redo' + 'automation/redo', + 'automation/inject-message' ] supportedActions.should.only.have.keys(...expectedKeys) }) @@ -5112,6 +5113,64 @@ describeMain('expertAutomations', () => { }) }) + describe('inject-message action', () => { + let node + beforeEach(() => { + node = { id: 'n1', type: 'function', z: 'tab-1', inputs: 1 } + mockRED.nodes.node.callsFake(id => id === 'n1' ? node : null) + mockRED.nodes.subflow = sinon.stub().returns(null) + global.$ = { ajax: sinon.stub().resolves({ status: 'ok', injected: true, msgid: 'm1' }) } + }) + afterEach(() => { + delete global.$ + }) + it('posts the msg to the runtime and returns the new msgid', async () => { + const result = {} + await expertAutomations.invokeAction('automation/inject-message', { params: { id: 'n1', msg: { payload: 42 } } }, result) + global.$.ajax.calledOnce.should.be.true() + const request = global.$.ajax.firstCall.args[0] + request.should.have.properties({ url: 'nr-assistant/inject', method: 'POST' }) + JSON.parse(request.data).should.eql({ id: 'n1', msg: { payload: 42 } }) + result.should.have.properties({ success: true, injected: true, msgid: 'm1' }) + }) + it('sends an empty msg when none is provided', async () => { + const result = {} + await expertAutomations.invokeAction('automation/inject-message', { params: { id: 'n1' } }, result) + JSON.parse(global.$.ajax.firstCall.args[0].data).should.eql({ id: 'n1', msg: {} }) + }) + it('reports injected: false when the runtime refuses (agent deploy not enabled)', async () => { + global.$.ajax.resolves({ status: 'ok', injected: false, code: 'INJECT_NOT_ALLOWED' }) + const result = {} + await expertAutomations.invokeAction('automation/inject-message', { params: { id: 'n1' } }, result) + result.should.have.properties({ success: true, injected: false, code: 'INJECT_NOT_ALLOWED' }) + result.should.not.have.property('msgid') + }) + it('reports the runtime error message when the request fails', async () => { + global.$.ajax.rejects({ responseJSON: { message: 'Node n1 not found in the deployed flows' } }) + const result = {} + await expertAutomations.invokeAction('automation/inject-message', { params: { id: 'n1' } }, result) + result.should.have.properties({ success: false, error: 'Node n1 not found in the deployed flows' }) + }) + const refusals = [ + ['an unknown node', () => { node = null }, /not found/], + ['a node with no input port', () => { node.inputs = 0 }, /no input port/], + ['a disabled node', () => { node.d = true }, /disabled/], + ['a node on a disabled tab', () => { mockRED.nodes.workspace.returns({ id: 'tab-1', disabled: true }) }, /disabled/], + ['a node inside a subflow definition', () => { mockRED.nodes.subflow.withArgs('tab-1').returns({ id: 'tab-1' }) }, /subflow/], + ['a node with undeployed changes', () => { node.changed = true }, /deploy first/] + ] + for (const [label, setup, error] of refusals) { + it(`refuses ${label} without calling the runtime`, async () => { + setup() + const result = {} + await expertAutomations.invokeAction('automation/inject-message', { params: { id: 'n1' } }, result) + result.should.have.property('success', false) + result.should.have.property('error').which.match(error) + global.$.ajax.called.should.be.false() + }) + } + }) + describe('export-flow action', () => { let clickedId beforeEach(() => {