Made mosquitto auth work

Dockerfile.mosquitto

@@ -0,0 +1,11 @@
+FROM debian:stretch
+
+LABEL Author="Alex Mikhalev"
+LABEL Description="Eclipse Mosquitto MQTT Broker"
+
+RUN apt-get update && \
+    apt-get install -y mosquitto mosquitto-auth-plugin
+
+COPY sprinklers3.mosquitto.conf /etc/mosquitto/conf.d/
+
+CMD ["/usr/sbin/mosquitto", "-c", "/etc/mosquitto/mosquitto.conf"]

client/state/AppState.ts

@@ -28,9 +28,9 @@ export default class AppState extends TypedEventEmitter<AppEvents> {
     constructor() {
         super();
         this.sprinklersRpc.on("newUserData", this.userStore.receiveUserData);
-        this.sprinklersRpc.on("tokenError", this.checkToken);
+        this.sprinklersRpc.on("tokenError", this.clearToken);
         this.httpApi.on("tokenGranted", () => this.emit("hasToken"));
-        this.httpApi.on("tokenError", this.checkToken);
+        this.httpApi.on("tokenError", this.clearToken);
 
         this.on("checkToken", this.doCheckToken);
 
@@ -55,6 +55,11 @@ export default class AppState extends TypedEventEmitter<AppEvents> {
         await this.checkToken();
     }
 
+    clearToken = (err?: any) => {
+        this.tokenStore.clearAccessToken();
+        this.checkToken();
+    }
+
     checkToken = () => {
         this.emit("checkToken");
     }
@@ -76,7 +81,7 @@ export default class AppState extends TypedEventEmitter<AppEvents> {
         } catch (err) {
             if (err instanceof ApiError && err.code === ErrorCode.BadToken) {
                 log.warn({ err }, "refresh is bad for some reason, erasing");
-                this.tokenStore.clear();
+                this.tokenStore.clearAll();
                 this.history.push("/login");
             } else {
                 log.error({ err }, "could not refresh access token");

client/state/TokenStore.ts

@@ -10,7 +10,13 @@ export class TokenStore {
     @observable refreshToken: Token<RefreshToken> = new Token();
 
     @action
-    clear() {
+    clearAccessToken() {
+        this.accessToken.token = null;
+        this.saveLocalStorage();
+    }
+
+    @action
+    clearAll() {
         this.accessToken.token = null;
         this.refreshToken.token = null;
         this.saveLocalStorage();

common/TokenClaims.ts

@@ -22,6 +22,11 @@ export interface DeviceRegistrationToken extends BaseClaims {
 export interface DeviceToken extends BaseClaims {
     type: "device";
     aud: string;
+    id: number;
 }
 
-export type TokenClaims = AccessToken | RefreshToken | DeviceRegistrationToken | DeviceToken;
+export interface SuperuserToken extends BaseClaims {
+    type: "superuser";
+}
+
+export type TokenClaims = AccessToken | RefreshToken | DeviceRegistrationToken | DeviceToken | SuperuserToken;

common/sprinklersRpc/mqtt/index.ts

@@ -17,7 +17,15 @@ interface WithRid {
     rid: number;
 }
 
-export class MqttRpcClient implements s.SprinklersRPC {
+export const DEVICE_PREFIX = "devices";
+
+export interface MqttRpcClientOptions {
+    mqttUri: string;
+    username?: string;
+    password?: string;
+}
+
+export class MqttRpcClient implements s.SprinklersRPC, MqttRpcClientOptions {
     get connected(): boolean {
         return this.connectionState.isServerConnected || false;
     }
@@ -26,21 +34,26 @@ export class MqttRpcClient implements s.SprinklersRPC {
         return "sprinklers3-MqttApiClient-" + getRandomId();
     }
 
-    readonly mqttUri: string;
+    mqttUri!: string;
+    username?: string;
+    password?: string;
+
     client!: mqtt.Client;
     @observable connectionState: s.ConnectionState = new s.ConnectionState();
     devices: Map<string, MqttSprinklersDevice> = new Map();
 
-    constructor(mqttUri: string) {
+    constructor(opts: MqttRpcClientOptions) {
-        this.mqttUri = mqttUri;
+        Object.assign(this, opts);
         this.connectionState.serverToBroker = false;
     }
 
     start() {
         const clientId = MqttRpcClient.newClientId();
-        log.info({ mqttUri: this.mqttUri, clientId }, "connecting to mqtt broker with client id");
+        const mqttUri = this.mqttUri;
-        this.client = mqtt.connect(this.mqttUri, {
+        log.info({ mqttUri, clientId }, "connecting to mqtt broker with client id");
+        this.client = mqtt.connect(mqttUri, {
             clientId, connectTimeout: 5000, reconnectPeriod: 5000,
+            username: this.username, password: this.password,
         });
         this.client.on("message", this.onMessageArrived.bind(this));
         this.client.on("close", () => {
@@ -90,12 +103,16 @@ export class MqttRpcClient implements s.SprinklersRPC {
     private processMessage(topic: string, payloadBuf: Buffer, packet: mqtt.Packet) {
         const payload = payloadBuf.toString("utf8");
         log.trace({ topic, payload }, "message arrived: ");
-        const topicIdx = topic.indexOf("/"); // find the first /
+        const regexp = new RegExp(`^${DEVICE_PREFIX}\\/([^\\/]+)\\/?(.*)$`);
-        const prefix = topic.substr(0, topicIdx); // assume prefix does not contain a /
+        const matches = regexp.exec(topic);
-        const topicSuffix = topic.substr(topicIdx + 1);
+        if (!matches) {
-        const device = this.devices.get(prefix);
+            return log.warn({ topic }, "received message on invalid topic");
+        }
+        const id = matches[1];
+        const topicSuffix = matches[2];
+        const device = this.devices.get(id);
         if (!device) {
-            log.debug({ prefix }, "received message for unknown device");
+            log.debug({ id }, "received message for unknown device");
             return;
         }
         device.onMessage(topicSuffix, payload);
@@ -131,20 +148,22 @@ const handler = (test: RegExp) =>
 
 class MqttSprinklersDevice extends s.SprinklersDevice {
     readonly apiClient: MqttRpcClient;
-    readonly prefix: string;
+    readonly id: string;
 
     handlers!: IHandlerEntry[];
+    private subscriptions: string[];
     private nextRequestId: number = Math.floor(Math.random() * 1000000000);
     private responseCallbacks: Map<number, ResponseCallback> = new Map();
 
-    constructor(apiClient: MqttRpcClient, prefix: string) {
+    constructor(apiClient: MqttRpcClient, id: string) {
         super();
         this.sectionConstructor = MqttSection;
         this.sectionRunnerConstructor = MqttSectionRunner;
         this.programConstructor = MqttProgram;
         this.apiClient = apiClient;
-        this.prefix = prefix;
+        this.id = id;
         this.sectionRunner = new MqttSectionRunner(this);
+        this.subscriptions = subscriptions.map((filter) => this.prefix + filter);
 
         autorun(() => {
             const brokerConnected = apiClient.connected;
@@ -160,14 +179,13 @@ class MqttSprinklersDevice extends s.SprinklersDevice {
         });
     }
 
-    get id(): string {
+    get prefix(): string {
-        return this.prefix;
+        return DEVICE_PREFIX + "/" + this.id;
     }
 
     doSubscribe(): Promise<void> {
-        const topics = subscriptions.map((filter) => this.prefix + filter);
         return new Promise((resolve, reject) => {
-            this.apiClient.client.subscribe(topics, { qos: 1 }, (err) => {
+            this.apiClient.client.subscribe(this.subscriptions, { qos: 1 }, (err) => {
                 if (err) {
                     reject(err);
                 } else {
@@ -178,9 +196,8 @@ class MqttSprinklersDevice extends s.SprinklersDevice {
     }
 
     doUnsubscribe(): Promise<void> {
-        const topics = subscriptions.map((filter) => this.prefix + filter);
         return new Promise((resolve, reject) => {
-            this.apiClient.client.unsubscribe(topics, (err) => {
+            this.apiClient.client.unsubscribe(this.subscriptions, (err) => {
                 if (err) {
                     reject(err);
                 } else {

docker-compose.dev.yml

@@ -7,6 +7,7 @@ services:
       dockerfile: Dockerfile.dev
     depends_on:
       - database
+      - mosquitto
     ports:
       - "8080:8080"
       - "8081:8081"
@@ -23,9 +24,19 @@ services:
       - TYPEORM_DATABASE=postgres
       - TYPEORM_USERNAME=postgres
       - TYPEORM_PASSWORD=8JN4w0UsN5dbjMjNvPe452P2yYOqg5PV
+      - MQTT_URL=tcp://mosquitto:1883
       # Must specify JWT_SECRET and MQTT_URL
 
+  mosquitto:
+    build:
+      context: .
+      dockerfile: Dockerfile.mosquitto
+    ports:
+      - "1883:1883"
+
   database:
     image: "postgres:11-alpine"
+    ports:
+      - "5432:5432"
     environment:
       - POSTGRES_PASSWORD=8JN4w0UsN5dbjMjNvPe452P2yYOqg5PV

server/express/api/devices.ts

@@ -49,7 +49,7 @@ export function devices(state: ServerState) {
             name: "Sprinklers Device", deviceId,
         });
         await state.database.sprinklersDevices.save(newDevice);
-        const token = await generateDeviceToken(deviceId);
+        const token = await generateDeviceToken(newDevice.id, deviceId);
         res.send({
             data: newDevice, token,
         });

server/express/api/mosquitto.ts

@@ -1,19 +1,56 @@
 import PromiseRouter from "express-promise-router";
 
+import ApiError from "@common/ApiError";
+import { ErrorCode } from "@common/ErrorCode";
+import { DEVICE_PREFIX } from "@common/sprinklersRpc/mqtt";
+import { DeviceToken, SuperuserToken } from "@common/TokenClaims";
+import { verifyToken } from "@server/express/authentication";
 import { ServerState } from "@server/state";
 
+export const SUPERUSER = "sprinklers3";
+
 export function mosquitto(state: ServerState) {
     const router = PromiseRouter();
 
     router.post("/auth", async (req, res) => {
-        res.status(200).send();
+        const body = req.body;
+        const { username, password, topic, acc } = body;
+        if (typeof username !== "string" || typeof password !== "string") {
+            throw new ApiError("Must specify a username and password", ErrorCode.BadRequest);
+        }
+        if (username === SUPERUSER) {
+            await verifyToken<SuperuserToken>(password, "superuser");
+            return res.status(200).send({ username });
+        }
+        const claims = await verifyToken<DeviceToken>(password, "device");
+        if (claims.aud !== username) {
+            throw new ApiError("Username does not match token", ErrorCode.BadRequest);
+        }
+        res.status(200).send({
+            username, id: claims.id,
+        });
     });
 
     router.post("/superuser", async (req, res) => {
+        const { username } = req.body;
+        if (typeof username !== "string") {
+            throw new ApiError("Must specify a username", ErrorCode.BadRequest);
+        }
+        if (username !== SUPERUSER) {
+            return res.status(403).send();
+        }
         res.status(200).send();
     });
 
     router.post("/acl", async (req, res) => {
+        const { username, topic, clientid, acc } = req.body;
+        if (typeof username !== "string" || typeof topic !== "string") {
+            throw new ApiError("username and topic must be specified as strings", ErrorCode.BadRequest);
+        }
+        const prefix = DEVICE_PREFIX + "/" + username;
+        if (!topic.startsWith(prefix)) {
+            throw new ApiError(`device ${username} cannot access topic ${topic}`);
+        }
         res.status(200).send();
     });
 

server/express/authentication.ts

@@ -10,7 +10,7 @@ import {
     TokenGrantRequest,
     TokenGrantResponse,
 } from "@common/httpApi";
-import { AccessToken, DeviceRegistrationToken, DeviceToken, RefreshToken, TokenClaims } from "@common/TokenClaims";
+import { AccessToken, DeviceRegistrationToken, DeviceToken, RefreshToken, TokenClaims, SuperuserToken } from "@common/TokenClaims";
 import { User } from "../entities";
 import { ServerState } from "../state";
 
@@ -110,15 +110,24 @@ function generateDeviceRegistrationToken(secret: string): Promise<string> {
     return signToken(device_reg_token_claims);
 }
 
-export function generateDeviceToken(deviceId: string): Promise<string> {
+export function generateDeviceToken(id: number, deviceId: string): Promise<string> {
     const device_token_claims: DeviceToken = {
         iss: ISSUER,
         type: "device",
         aud: deviceId,
+        id,
     };
     return signToken(device_token_claims);
 }
 
+export function generateSuperuserToken(): Promise<string> {
+    const superuser_claims: SuperuserToken = {
+        iss: ISSUER,
+        type: "superuser",
+    };
+    return signToken(superuser_claims);
+}
+
 export function authentication(state: ServerState) {
 
     const router = Router();
@@ -143,15 +152,15 @@ export function authentication(state: ServerState) {
     async function refreshGrant(body: TokenGrantRefreshRequest, res: Express.Response): Promise<User> {
         const { refresh_token } = body;
         if (!body || !refresh_token) {
-            throw new ApiError("Must specify a refresh_token");
+            throw new ApiError("Must specify a refresh_token", ErrorCode.BadToken);
         }
         const claims = await verifyToken(refresh_token);
         if (claims.type !== "refresh") {
-            throw new ApiError("Not a refresh token");
+            throw new ApiError("Not a refresh token", ErrorCode.BadToken);
         }
         const user = await state.database.users.findOne(claims.aud);
         if (!user) {
-            throw new ApiError("User no longer exists");
+            throw new ApiError("User no longer exists", ErrorCode.BadToken);
         }
         return user;
     }

server/sprinklersRpc/websocketServer.ts

@@ -88,11 +88,14 @@ export class WebSocketClient {
             this.userId = claims.aud;
             this.user = await this.state.database.users.
                 findById(this.userId, { devices: true }) || null;
+            if (!this.user) {
+                throw new ws.RpcError("user no longer exists", ErrorCode.BadToken);
+            }
             log.info({ userId: claims.aud, name: claims.name }, "authenticated websocket client");
             this.subscribeBrokerConnection();
             return {
                 result: "success",
-                data: { authenticated: true, message: "authenticated", user: this.user!.toJSON() },
+                data: { authenticated: true, message: "authenticated", user: this.user.toJSON() },
             };
         },
         deviceSubscribe: async (data: ws.IDeviceSubscribeRequest) => {

server/state.ts

@@ -1,5 +1,7 @@
 import logger from "@common/logger";
 import * as mqtt from "@common/sprinklersRpc/mqtt";
+import { SUPERUSER } from "@server/express/api/mosquitto";
+import { generateSuperuserToken } from "@server/express/authentication";
 import { Database } from "./Database";
 
 export class ServerState {
@@ -13,7 +15,9 @@ export class ServerState {
             throw new Error("Must specify a MQTT_URL to connect to");
         }
         this.mqttUrl = mqttUrl;
-        this.mqttClient = new mqtt.MqttRpcClient(mqttUrl);
+        this.mqttClient = new mqtt.MqttRpcClient({
+            mqttUri: mqttUrl,
+        });
         this.database = new Database();
     }
 
@@ -22,6 +26,8 @@ export class ServerState {
         await this.database.createAll();
         logger.info("created database and tables");
 
+        this.mqttClient.username = SUPERUSER;
+        this.mqttClient.password = await generateSuperuserToken();
         this.mqttClient.start();
     }
 }

sprinklers3.mosquitto.conf

@@ -0,0 +1,8 @@
+allow_anonymous false
+auth_plugin /usr/lib/mosquitto-auth-plugin/auth-plugin.so
+auth_opt_backends http
+auth_opt_http_ip web
+auth_opt_http_port 8080
+auth_opt_http_getuser_uri /api/mosquitto/auth
+auth_opt_http_superuser_uri /api/mosquitto/superuser
+auth_opt_http_aclcheck_uri /api/mosquitto/acl