Team broker schema agent

forge/comms/v2AuthRoutes.js

@@ -52,6 +52,23 @@ module.exports = async function (app) {
                     result: 'deny'
                 })
             }
+        } else if (username.startsWith('agent:')) {
+            const parts = username.split('@')
+            const teamId = parts[1]
+            const agent = await app.db.models.TeamBrokerAgent.byTeam(teamId)
+            if (agent && agent.auth === password) {
+                reply.send({
+                    result: 'allow',
+                    is_superuser: false,
+                    client_attrs: {
+                        team: `ff/v1/${teamId}/c/`
+                    }
+                })
+            } else {
+                reply.send({
+                    result: 'deny'
+                })
+            }
         } else {
             if (app.license.active()) {
                 let teamId = null
@@ -134,6 +151,12 @@ module.exports = async function (app) {
                 reply.send({ result: 'deny' })
             }
             // return
+        } else if (username.startsWith('agent:')) {
+            if (action === 'subscribe') {
+                reply.send({ result: 'allow' })
+            } else {
+                reply.send({ result: 'deny' })
+            }
         } else {
             if (app.license.active()) {
                 let teamClientUsername

forge/db/controllers/AccessToken.js

@@ -437,6 +437,27 @@ module.exports = {
         return { token }
     },
 
+    createTokenForTeamBrokerAgent: async function (app, broker, expiresAt, scope = ['broker:credentials', 'broker:credentials:edit', 'broker:topics']) {
+        const existingBrokerToken = await app.db.models.AccessToken.findOne({
+            where: {
+                ownerId: '' + broker.id,
+                ownerType: 'teamBrokerAgent'
+            }
+        })
+        if (existingBrokerToken) {
+            await existingBrokerToken.destroy()
+        }
+        const token = generateToken(32, 'fftpb')
+        await app.db.models.AccessToken.create({
+            token,
+            expiresAt,
+            scope,
+            ownerId: '' + broker.id,
+            ownerType: 'teamBrokerAgent'
+        })
+        return { token }
+    },
+
     createTokenForNPM: async function (app, entity, team, scope = ['team:packages:read']) {
         // Adding prefix to the entityId of `p-`, `d-` and `u-` rather than relying on
         // no hashid collisions

forge/db/migrations/20250908-01-EE-add-team-broker-agent.js

@@ -0,0 +1,24 @@
+/**
+ * Add TeamBrokerAgents table
+ */
+const { DataTypes } = require('sequelize')
+
+module.exports = {
+    up: async (context) => {
+        await context.createTable('TeamBrokerAgents', {
+            id: { type: DataTypes.INTEGER, primaryKey: true, autoIncrement: true },
+            state: { type: DataTypes.STRING, allowNull: false, default: 'running' },
+            settings: { type: DataTypes.TEXT, allowNull: true },
+            auth: { type: DataTypes.STRING, allowNull: false },
+            createdAt: { type: DataTypes.DATE, allowNull: false },
+            updatedAt: { type: DataTypes.DATE, allowNull: false },
+            TeamId: {
+                type: DataTypes.INTEGER,
+                references: { model: 'Teams', key: 'id' },
+                onDelete: 'CASCADE',
+                onUpdate: 'CASCADE'
+            }
+        })
+    },
+    down: async (context) => {}
+}

forge/db/models/TeamBrokerAgent.js

@@ -0,0 +1,101 @@
+/**
+ * TeamBrokerAgent table
+ */
+const { DataTypes } = require('sequelize')
+
+const { generatePassword } = require('../../lib/userTeam')
+const Controllers = require('../controllers')
+
+module.exports = {
+    name: 'TeamBrokerAgent',
+    schema: {
+        state: { type: DataTypes.STRING, allowNull: false, default: 'running' },
+        settings: {
+            type: DataTypes.TEXT,
+            allowNull: true,
+            default: '{}',
+            get () {
+                const rawValue = this.getDataValue('settings')
+                if (rawValue) {
+                    return JSON.parse(rawValue)
+                } else {
+                    return {}
+                }
+            },
+            set (value) {
+                if (value) {
+                    this.setDataValue('settings', JSON.stringify(value))
+                }
+            }
+        },
+        auth: {
+            type: DataTypes.STRING,
+            allowNull: false,
+            defaultValue: function () {
+                return generatePassword()
+            }
+        }
+    },
+    indexes: [
+        { name: 'team_broker_agent_team_unique', fields: ['id', 'TeamId'], unique: true }
+    ],
+    hooks: function (M, app) {
+        return {
+            afterDestroy: async (teamBrokerAgent, opts) => {
+                await M.AccessToken.destroy({
+                    where: {
+                        ownerType: 'teamBrokerAgent',
+                        ownerId: '' + teamBrokerAgent.id
+                    }
+                })
+            }
+        }
+    },
+    associations: function (M) {
+        this.belongsTo(M.Team)
+        this.hasOne(M.AccessToken, {
+            foreignKey: 'ownerId',
+            constraints: false,
+            scope: {
+                ownerType: 'teamBrokerAgent'
+            }
+        })
+    },
+    finders: function (M) {
+        return {
+            instance: {
+                async refreshAuthTokens () {
+                    const brokerToken = await Controllers.AccessToken.createTokenForTeamBrokerAgent(this, null)
+                    return {
+                        token: brokerToken.token
+                    }
+                }
+            },
+            static: {
+                byId: async function (idOrHash) {
+                    let id = idOrHash
+                    if (typeof id === 'string') {
+                        id = M.TeamBrokerAgent.decodeHashid(idOrHash)
+                    }
+                    return this.findOne({
+                        where: { id },
+                        include: {
+                            model: M.Team
+                        }
+                    })
+                },
+                byTeam: async function (teamId) {
+                    if (typeof teamId === 'string') {
+                        teamId = M.Team.decodeHashid(teamId)
+                    }
+                    return this.findOne({
+                        include: {
+                            model: M.Team,
+                            where: { id: teamId }
+                        }
+                    })
+                }
+            }
+        }
+    }
+}

forge/db/models/index.js

@@ -82,7 +82,8 @@ const modelTypes = [
     'Notification',
     'TeamBrokerClient',
     'BrokerCredentials',
-    'MQTTTopicSchema'
+    'MQTTTopicSchema',
+    'TeamBrokerAgent'
 ]
 
 // A local map of the known models.

forge/ee/routes/teamBroker/3rdPartyBroker.js

@@ -160,6 +160,10 @@ module.exports = async function (app) {
                         if (request.params.teamId !== request.session.Broker.Team.hashid) {
                             reply.code('401').send({ code: 'unauthorized', error: 'unauthorized' })
                         }
+                    } else if (request.session.ownerType === 'teamBrokerAgent') {
+                        if (request.params.teamId !== request.session.TeamBrokerAgent.Team.hashid) {
+                            reply.code('401').send({ code: 'unauthorized', error: 'unauthorized' })
+                        }
                     } else {
                         reply.code('401').send({ code: 'unauthorized', error: 'unauthorized' })
                     }
@@ -207,6 +211,28 @@ module.exports = async function (app) {
             } else {
                 reply.code('401').send({ code: 'unauthorized', error: 'unauthorized' })
             }
+        } else if (request.params.brokerId === 'team-broker') {
+            // provide team level creds for team broker
+            const agent = await app.db.models.TeamBrokerAgent.byTeam(request.params.teamId)
+            if (agent) {
+                const brokerURL = new URL(app.config.broker.url)
+                reply.send({
+                    id: 'team-broker',
+                    name: 'TeamBroker',
+                    host: brokerURL.hostname,
+                    port: brokerURL.port,
+                    protocol: brokerURL.protocol,
+                    protocolVersion: '4',
+                    ssl: brokerURL.protocol === 'mqtts:' || brokerURL.protocol === 'wss:',
+                    verifySSL: true,
+                    clientId: `${request.params.teamId}-agent@${request.params.teamId}`,
+                    credentials: {
+                        username: `agent:${request.params.teamId}@${request.params.teamId}`,
+                        password: agent.auth
+                    },
+                    topicPrefix: ['#']
+                })
+            }
         } else {
             reply.status(404).send({ code: 'not_found', error: 'not found' })
         }
@@ -316,7 +342,20 @@ module.exports = async function (app) {
                 reply.status(500).send({ error: 'unknown_error', message: err.toString() })
             }
         } else {
-            reply.status(400).send({ error: 'not_supported', message: 'not supported' })
+            const agentStatus = await app.db.models.TeamBrokerAgent.byTeam(request.params.teamId)
+            if (agentStatus) {
+                const clean = agentStatus.toJSON()
+                delete clean.auth
+                clean.status = {
+                    connected: true,
+                    error: null
+                }
+                reply.send(clean)
+            } else {
+                reply.send({
+                    state: 'suspended'
+                })
+            }
         }
     })
 
@@ -381,7 +420,19 @@ module.exports = async function (app) {
         schema: { }
     }, async (request, reply) => {
         if (request.params.brokerId === 'team-broker') {
-            reply.status(403).send({})
+            const agent = await app.db.models.TeamBrokerAgent.byTeam(request.team.hashid)
+            if (agent && agent.state !== 'running') {
+                await app.containers.sendBrokerAgentCommand(agent, 'start')
+                reply.status(200).send({})
+            } else if (!agent) {
+                const agent = await app.db.models.TeamBrokerAgent.create({
+                    state: 'running',
+                    TeamId: request.team.id
+                })
+                await agent.reload({ include: [{ model: app.db.models.Team }] })
+                await app.containers.startBrokerAgent(agent)
+                reply.status(200).send({})
+            }
         } else {
             if (request.broker.state === 'running') {
                 await app.containers.sendBrokerAgentCommand(request.broker, 'start')
@@ -405,7 +456,14 @@ module.exports = async function (app) {
         schema: { }
     }, async (request, reply) => {
         if (request.params.brokerId === 'team-broker') {
-            reply.status(403).send({})
+            const agent = await app.db.models.TeamBrokerAgent.byTeam(request.team.hashid)
+            if (agent && agent.state === 'running') {
+                await app.containers.sendBrokerAgentCommand(agent, 'stop')
+                reply.status(200).send({})
+            } else {
+                // hmm shouldn't be able to get here
+            }
+            // reply.status(403).send({})
         } else {
             await app.containers.sendBrokerAgentCommand(request.broker, 'stop')
             reply.status(200).send({})
@@ -423,7 +481,16 @@ module.exports = async function (app) {
         schema: { }
     }, async (request, reply) => {
         if (request.params.brokerId === 'team-broker') {
-            reply.status(403).send({})
+            const agent = await app.db.models.TeamBrokerAgent.byTeam(request.team.id)
+            if (agent) {
+                await app.containers.stopBrokerAgent(agent)
+                setTimeout(async () => {
+                    await agent.destroy()
+                }, 1500)
+                reply.status(200).send({})
+            } else {
+                reply.status(404).send({})
+            }
         } else {
             await app.containers.stopBrokerAgent(request.broker)
             request.broker.state = 'suspended'
@@ -493,6 +560,10 @@ module.exports = async function (app) {
                         if (request.params.teamId !== request.session.Broker.Team.hashid) {
                             reply.code('401').send({ code: 'unauthorized', error: 'unauthorized' })
                         }
+                    } else if (request.session.ownerType === 'teamBrokerAgent') {
+                        if (request.params.teamId !== request.session.TeamBrokerAgent.Team.hashid) {
+                            reply.code('401').send({ code: 'unauthorized', error: 'unauthorized' })
+                        }
                     } else {
                         reply.code('401').send({ code: 'unauthorized', error: 'unauthorized' })
                     }

forge/routes/auth/index.js

@@ -131,6 +131,13 @@ async function init (app, opts) {
                             return
                         }
                     }
+                    if (accessToken.ownerType === 'teamBrokerAgent') {
+                        request.session.TeamBrokerAgent = await app.db.models.TeamBrokerAgent.byId(parseInt(accessToken.ownerId))
+                        if (!request.session.TeamBrokerAgent) {
+                            reply.code(401).send({ code: 'unauthorized', error: 'unauthorized' })
+                            return
+                        }
+                    }
                     return
                 }
                 reply.code(401).send({ code: 'unauthorized', error: 'unauthorized' })