MQTT Integration

This document covers MQTT authorization, topic patterns, Rego policies, and broker configuration.

Overview

The CELINE MQTT setup consists of:

Mosquitto with mosquitto-go-auth plugin
MQTT auth service (celine.mqtt_auth) as the HTTP backend
Rego policies (policies/celine/mqtt/acl.rego + policies/celine/scopes.rego) evaluated via regorus

MQTT clients authenticate with a JWT (obtained from Keycloak) passed as the MQTT password.

Authentication Flow

Client connects to Mosquitto with JWT as password
Mosquitto calls POST /user on the auth service
Auth service validates JWT signature, issuer, and expiry
On publish/subscribe, Mosquitto calls POST /acl
Auth service evaluates the celine.mqtt.acl Rego policy
Superuser check via POST /superuser is disabled in the mosquitto config (auth_opt_disable_superuser true), and the endpoint answers 403 to every token anyway (REQ-0014)

Topic Naming Convention

celine/{service}/{resource}/{...}

The ACL policy parses topics by splitting on / and derives the required scope as {service}.{resource}.{verb}.

Examples:

Topic	Action	Required Scope
`celine/pipelines/runs/pipeline-123`	subscribe	`pipelines.runs.read`
`celine/digital-twin/events/pump/pump-001`	publish	`digital-twin.events.write`
`celine/flexibility/committed/flex-456`	subscribe	`flexibility.committed.read`
`celine/nudging/ingest/user-789`	publish	`nudging.ingest.write`

ACL Policy Rules

The policy (policies/celine/mqtt/acl.rego) evaluates access based on topic shape:

Service-level topics (`celine/{service}`)

Access to celine/{service} (no resource path) requires the service admin scope ({service}.admin), held by a service account.

Service wildcard topics (`celine/{service}/#` or `celine/{service}/+`)

Same requirement as service-level topics — only a service admin scope opens a service-wide wildcard.

Resource topics (`celine/{service}/{resource}/{...}`)

A service account needs one of: - Exact scope match ({service}.{resource}.{verb}) - Service admin scope ({service}.admin) - Resource wildcard scope ({service}.{resource}.*)

A person's token reaches no topic (REQ-0014, ADR-0012). User or service is the token's kind (is_service_account), never whether it holds a group; the policy input carries no group of either level, so a realm group, an organization group or a group named like a grant (admin, mqtt.admin, {service}.admin, {service}.{resource}.{verb}, mqtt:...) grants nothing, and neither do the scopes a browser token carries or the realm role platform-admin.

Action mapping

MQTT action	Rego verb
`subscribe`	`read`
`read`	`read`
`publish`	`write`

Mosquitto Configuration

The broker config is at config/mosquitto/mosquitto.conf:

listener 1883           # MQTT
listener 1884           # WebSockets
protocol mqtt / websockets

auth_plugin /mosquitto/go-auth.so
auth_opt_backends jwt
auth_opt_jwt_mode remote
auth_opt_jwt_host host.docker.internal
auth_opt_jwt_port 8009

auth_opt_jwt_getuser_uri    /user
auth_opt_jwt_aclcheck_uri   /acl
auth_opt_jwt_superuser_uri  /superuser

auth_opt_disable_superuser true

# Redis caching (disabled by default, redis is available in compose)
auth_opt_cache false
auth_opt_cache_type redis
auth_opt_cache_host host.docker.internal
auth_opt_cache_port 6379

The broker uses the JWT backend mode (auth_opt_backends jwt), not the HTTP backend. The JWT is extracted by mosquitto-go-auth and forwarded to the auth service endpoints.

Docker Compose

The relevant services in docker-compose.yaml:

mqtt_auth:          # FastAPI auth service on port 8009
mosquitto:          # Mosquitto broker on ports 1883 (MQTT) + 1884 (WS)
redis:              # Redis for optional auth caching

Mosquitto depends on mqtt_auth being healthy before starting.

Client Examples

Python (paho-mqtt)

import paho.mqtt.client as mqtt

token = get_jwt_from_keycloak()

client = mqtt.Client()
client.username_pw_set(username="", password=token)
client.connect("localhost", 1883)

# Subscribe (requires the {service}.{resource}.read scope, on a service account)
client.subscribe("celine/digital-twin/events/#")

# Publish (requires the {service}.{resource}.write scope, on a service account)
client.publish("celine/digital-twin/events/pump/pump-001", payload='{"state": "running"}')

Service Account

Service clients use client_credentials flow to get a JWT with the scopes defined in clients.yaml:

import httpx

response = httpx.post(
    "http://keycloak.celine.localhost/realms/celine/protocol/openid-connect/token",
    data={
        "grant_type": "client_credentials",
        "client_id": "svc-digital-twin",
        "client_secret": "your-secret",
    },
)
token = response.json()["access_token"]

Debugging

Test authentication

TOKEN=$(curl -s -X POST \
  "http://localhost:8080/realms/celine/protocol/openid-connect/token" \
  -d "grant_type=client_credentials" \
  -d "client_id=svc-digital-twin" \
  -d "client_secret=svc-digital-twin" \
  | jq -r '.access_token')

curl -X POST http://localhost:8009/user \
  -H "Authorization: Bearer $TOKEN"

Test ACL

curl -X POST http://localhost:8009/acl \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"clientid": "test", "topic": "celine/digital-twin/events/pump/1", "acc": 2}'

Check health

curl http://localhost:8009/health

Decode JWT claims

echo $TOKEN | cut -d. -f2 | base64 -d 2>/dev/null | jq