Secure socket broker with token auth and rooms

Dockers/socket-broker/index.js

@@ -1,278 +1,510 @@
+const crypto = require('crypto');
 const express = require('express');
 const { createServer } = require('http');
 const { Server } = require('socket.io');
 const cors = require('cors');
 
-// Create Express app and HTTP server
-const app = express();
-const server = createServer(app);
+const DEFAULT_ALLOWED_ORIGINS = [
+    'https://*.comedykit.be',
+    'https://comedykit.be',
+    'https://www.comedykit.be',
+    'http://localhost:5173',
+    'http://localhost:3000',
+    'capacitor://localhost',
+    'http://localhost',
+    'https://localhost',
+];
 
-// Create Socket.IO server
-const io = new Server(server, {
-    cors: {
-        origin: "*", // Allow all origins for development
-        methods: ["GET", "POST"]
+const DEFAULT_API_KEY = 'default-api-key-change-me';
+
+function parseConfig(env = process.env) {
+    const authMode = (env.AUTH_MODE || 'required').toLowerCase();
+
+    if (!['off', 'optional', 'required'].includes(authMode)) {
+        throw new Error(`Invalid AUTH_MODE "${authMode}". Use off, optional, or required.`);
     }
-});
 
-// Store topic subscriptions
-const topicSubscriptions = new Map(); // topic -> Set of socket IDs
-const socketTopics = new Map(); // socket ID -> Set of topics
+    const tokenSecret = env.SOCKET_TOKEN_SECRET || '';
 
-// Get API key from environment variable
-const API_KEY = process.env.API_KEY || 'default-api-key-change-me';
+    if (authMode === 'required' && !tokenSecret) {
+        throw new Error('SOCKET_TOKEN_SECRET is required when AUTH_MODE=required.');
+    }
 
-// API Key authentication middleware
-const authenticateApiKey = (req, res, next) => {
-    const authHeader = req.headers.authorization;
-    const apiKey = req.headers['x-api-key'] || req.query.api_key;
+    return {
+        wsPort: parseInt(env.WS_PORT || env.PORT || '3000', 10),
+        apiPort: parseInt(env.API_PORT || env.WS_PORT || env.PORT || '3000', 10),
+        apiKey: env.API_KEY || DEFAULT_API_KEY,
+        tokenSecret,
+        authMode,
+        maxTopicsPerSocket: parseInt(env.MAX_TOPICS_PER_SOCKET || '20', 10),
+        subscribeRateLimit: parseInt(env.SUBSCRIBE_RATE_LIMIT || '30', 10),
+        allowedOrigins: (env.ALLOWED_ORIGINS || DEFAULT_ALLOWED_ORIGINS.join(','))
+            .split(',')
+            .map(origin => origin.trim())
+            .filter(Boolean),
+    };
+}
+
+function isOriginAllowed(origin, allowedOrigins) {
+    if (!origin) {
+        return true;
+    }
+
+    return allowedOrigins.some(allowed => {
+        if (allowed === '*') {
+            return true;
+        }
+
+        if (allowed.includes('*.')) {
+            const [prefix, suffix] = allowed.split('*.');
+
+            return origin.startsWith(prefix) && origin.endsWith(`.${suffix}`);
+        }
+
+        return origin === allowed;
+    });
+}
+
+function createBroker(env = process.env) {
+    const config = parseConfig(env);
+    const subscribeRateWindowMs = 10_000;
+
+    const corsOptions = {
+        origin: (origin, callback) => {
+            if (isOriginAllowed(origin, config.allowedOrigins)) {
+                return callback(null, true);
+            }
+
+            return callback(new Error(`Origin ${origin} is not allowed`));
+        },
+        methods: ['GET', 'POST'],
+    };
+
+    const app = express();
+    const wsServer = createServer(app);
+    const io = new Server(wsServer, {
+        cors: corsOptions,
+        maxHttpBufferSize: 1e6,
+    });
+
+    app.use(cors(corsOptions));
+    app.use(express.json({ limit: '1mb' }));
+
+    const authenticateApiKey = (req, res, next) => {
+        const authHeader = req.headers.authorization;
+        const apiKey = req.headers['x-api-key'] || req.query.api_key;
+
+        if (!authHeader && !apiKey) {
+            return res.status(401).json({
+                error: 'Unauthorized',
+                message: 'API key is required.',
+            });
+        }
+
+        if (authHeader && authHeader.startsWith('Bearer ')) {
+            const token = authHeader.substring(7);
+
+            if (token === config.apiKey) {
+                return next();
+            }
+        }
+
+        if (apiKey && apiKey === config.apiKey) {
+            return next();
+        }
 
-    if (!authHeader && !apiKey) {
         return res.status(401).json({
             error: 'Unauthorized',
-            message: 'API key is required. Use Authorization header or x-api-key header or api_key query parameter.'
+            message: 'Invalid API key',
+        });
+    };
+
+    function verifyToken(token, expectedType) {
+        if (!config.tokenSecret || typeof token !== 'string') {
+            return null;
+        }
+
+        const parts = token.split('.');
+
+        if (parts.length !== 2) {
+            return null;
+        }
+
+        const [encodedPayload, encodedSignature] = parts;
+        const expectedSignature = crypto
+            .createHmac('sha256', config.tokenSecret)
+            .update(encodedPayload)
+            .digest('base64url');
+
+        const received = Buffer.from(encodedSignature);
+        const expected = Buffer.from(expectedSignature);
+
+        if (received.length !== expected.length || !crypto.timingSafeEqual(received, expected)) {
+            return null;
+        }
+
+        let payload;
+
+        try {
+            payload = JSON.parse(Buffer.from(encodedPayload, 'base64url').toString('utf8'));
+        } catch {
+            return null;
+        }
+
+        if (!payload || payload.v !== 1 || payload.typ !== expectedType) {
+            return null;
+        }
+
+        if (typeof payload.sub !== 'string' || typeof payload.exp !== 'number') {
+            return null;
+        }
+
+        // Allow a small clock skew between Laravel and the broker.
+        if (payload.exp + 60 < Math.floor(Date.now() / 1000)) {
+            return null;
+        }
+
+        return payload;
+    }
+
+    function isValidTopic(topic) {
+        if (typeof topic !== 'string' || topic.length === 0 || topic.length > 255) {
+            return false;
+        }
+
+        return /^(user\/[0-9a-fA-F-]{36}\/notifications|event\/[A-Za-z0-9_-]+\/ticket_statistics)$/.test(topic);
+    }
+
+    function normalizeSubscription(payload) {
+        if (typeof payload === 'string') {
+            return { topic: payload, token: null };
+        }
+
+        if (payload && typeof payload === 'object') {
+            return {
+                topic: payload.topic,
+                token: typeof payload.token === 'string' ? payload.token : null,
+            };
+        }
+
+        return { topic: null, token: null };
+    }
+
+    function canSubscribe(socket, topic, topicToken) {
+        if (config.authMode === 'off') {
+            return true;
+        }
+
+        if (!socket.data.auth) {
+            return config.authMode === 'optional';
+        }
+
+        const userTopic = topic.match(/^user\/([^/]+)\/notifications$/);
+
+        if (userTopic) {
+            return userTopic[1] === socket.data.auth.sub;
+        }
+
+        const verifiedTopicToken = verifyToken(topicToken, 'topic');
+
+        if (!verifiedTopicToken) {
+            return false;
+        }
+
+        return verifiedTopicToken.sub === socket.data.auth.sub && verifiedTopicToken.topic === topic;
+    }
+
+    function consumeSubscribeAttempt(socket) {
+        const now = Date.now();
+        const rate = socket.data.rate || { windowStart: now, count: 0 };
+
+        if (now - rate.windowStart > subscribeRateWindowMs) {
+            rate.windowStart = now;
+            rate.count = 0;
+        }
+
+        rate.count += 1;
+        socket.data.rate = rate;
+
+        return rate.count <= config.subscribeRateLimit;
+    }
+
+    function topicRoomEntries() {
+        return Array.from(io.sockets.adapter.rooms.entries()).filter(([room]) => !io.sockets.sockets.has(room));
+    }
+
+    function subscriberCount(topic) {
+        return io.sockets.adapter.rooms.get(topic)?.size || 0;
+    }
+
+    io.use((socket, next) => {
+        const token = socket.handshake.auth ? socket.handshake.auth.token : null;
+        const payload = verifyToken(token, 'connect');
+
+        if (payload) {
+            socket.data.auth = payload;
+            return next();
+        }
+
+        if (config.authMode === 'off' || config.authMode === 'optional') {
+            socket.data.auth = null;
+            return next();
+        }
+
+        return next(new Error('unauthorized'));
+    });
+
+    io.on('connection', socket => {
+        socket.data.topics = new Set();
+        console.log(`Client connected: ${socket.id}${socket.data.auth ? ` (user ${socket.data.auth.sub})` : ''}`);
+
+        socket.on('subscribe', payload => {
+            const { topic, token } = normalizeSubscription(payload);
+
+            if (!consumeSubscribeAttempt(socket)) {
+                socket.emit('error', { code: 'RATE_LIMITED', message: 'Too many subscription attempts' });
+                return;
+            }
+
+            if (!isValidTopic(topic)) {
+                socket.emit('error', { code: 'INVALID_TOPIC', message: 'Invalid topic' });
+                return;
+            }
+
+            if (!canSubscribe(socket, topic, token)) {
+                socket.emit('error', { code: 'FORBIDDEN_TOPIC', message: 'Not authorized for this topic', topic });
+                return;
+            }
+
+            if (socket.data.topics.has(topic)) {
+                socket.emit('subscribed', { topic, message: `Already subscribed to ${topic}` });
+                return;
+            }
+
+            if (socket.data.topics.size >= config.maxTopicsPerSocket) {
+                socket.emit('error', { code: 'TOPIC_LIMIT', message: 'Too many topic subscriptions' });
+                return;
+            }
+
+            socket.join(topic);
+            socket.data.topics.add(topic);
+
+            console.log(`Socket ${socket.id} subscribed to topic: ${topic}`);
+            socket.emit('subscribed', { topic, message: `Successfully subscribed to ${topic}` });
+        });
+
+        socket.on('unsubscribe', payload => {
+            const { topic } = normalizeSubscription(payload);
+
+            if (!isValidTopic(topic)) {
+                socket.emit('error', { code: 'INVALID_TOPIC', message: 'Invalid topic' });
+                return;
+            }
+
+            socket.leave(topic);
+            socket.data.topics.delete(topic);
+
+            console.log(`Socket ${socket.id} unsubscribed from topic: ${topic}`);
+            socket.emit('unsubscribed', { topic, message: `Successfully unsubscribed from ${topic}` });
+        });
+
+        socket.on('disconnect', () => {
+            console.log(`Client disconnected: ${socket.id}`);
+        });
+    });
+
+    app.get('/health', (req, res) => {
+        res.json({
+            status: 'healthy',
+            connections: io.engine.clientsCount,
+            timestamp: new Date().toISOString(),
+            apiKeyConfigured: !!config.apiKey && config.apiKey !== DEFAULT_API_KEY,
+            authMode: config.authMode,
+            tokenAuthConfigured: !!config.tokenSecret,
+        });
+    });
+
+    app.get('/topics', authenticateApiKey, (req, res) => {
+        const topics = topicRoomEntries().map(([topic, sockets]) => ({
+            topic,
+            subscribers: sockets.size,
+        }));
+
+        res.json({ topics });
+    });
+
+    app.get('/topics/*', authenticateApiKey, (req, res) => {
+        const topic = req.params[0];
+        const subscribers = io.sockets.adapter.rooms.get(topic)?.size;
+
+        if (!subscribers) {
+            return res.status(404).json({ error: 'Topic not found' });
+        }
+
+        return res.json({ topic, subscribers });
+    });
+
+    app.post('/publish', authenticateApiKey, (req, res) => {
+        const { topic, message, data } = req.body;
+
+        if (!topic || typeof topic !== 'string') {
+            return res.status(400).json({ error: 'Topic is required and must be a string' });
+        }
+
+        if (!message && !data) {
+            return res.status(400).json({ error: 'Either message or data is required' });
+        }
+
+        const subscribers = subscriberCount(topic);
+
+        if (subscribers === 0) {
+            return res.status(204).send();
+        }
+
+        const payload = {
+            topic,
+            message: message || null,
+            data: data || null,
+            timestamp: new Date().toISOString(),
+        };
+
+        io.to(topic).emit('message', payload);
+
+        console.log(`Message published to topic '${topic}' for ${subscribers} subscribers`);
+
+        return res.json({
+            success: true,
+            message: `Message published to topic '${topic}'`,
+            subscribers,
+            payload,
+        });
+    });
+
+    app.post('/publish/:topic', authenticateApiKey, (req, res) => {
+        const { topic } = req.params;
+        const { message, data } = req.body;
+
+        if (!message && !data) {
+            return res.status(400).json({ error: 'Either message or data is required' });
+        }
+
+        const subscribers = subscriberCount(topic);
+
+        if (subscribers === 0) {
+            return res.status(204).send();
+        }
+
+        const payload = {
+            topic,
+            message: message || null,
+            data: data || null,
+            timestamp: new Date().toISOString(),
+        };
+
+        io.to(topic).emit('message', payload);
+
+        console.log(`Message published to topic '${topic}' for ${subscribers} subscribers`);
+
+        return res.json({
+            success: true,
+            message: `Message published to topic '${topic}'`,
+            subscribers,
+            payload,
+        });
+    });
+
+    const apiServer = config.apiPort !== config.wsPort ? createServer(app) : null;
+    const servers = apiServer ? [wsServer, apiServer] : [wsServer];
+    let started = false;
+
+    function listen(server, port) {
+        return new Promise((resolve, reject) => {
+            const onError = error => {
+                server.removeListener('listening', onListening);
+                reject(error);
+            };
+
+            const onListening = () => {
+                server.removeListener('error', onError);
+                resolve(server.address().port);
+            };
+
+            server.once('error', onError);
+            server.once('listening', onListening);
+            server.listen(port);
         });
     }
 
-    // Check Authorization header (Bearer token format)
-    if (authHeader && authHeader.startsWith('Bearer ')) {
-        const token = authHeader.substring(7);
-        if (token === API_KEY) {
-            return next();
-        }
-    }
+    return {
+        config,
+        app,
+        io,
+        async start() {
+            const wsPort = await listen(wsServer, config.wsPort);
+            const apiPort = apiServer ? await listen(apiServer, config.apiPort) : wsPort;
+            started = true;
 
-    // Check x-api-key header
-    if (apiKey && apiKey === API_KEY) {
-        return next();
-    }
-
-    return res.status(401).json({
-        error: 'Unauthorized',
-        message: 'Invalid API key'
-    });
-};
-
-// Middleware
-app.use(cors());
-app.use(express.json());
-
-// WebSocket connection handling (Port 3000)
-io.on('connection', (socket) => {
-    console.log(`Client connected: ${socket.id}`);
-
-    // Handle topic subscription
-    socket.on('subscribe', (topic) => {
-        if (!topic || typeof topic !== 'string') {
-            socket.emit('error', { message: 'Invalid topic' });
-            return;
-        }
-
-        // Add socket to topic
-        if (!topicSubscriptions.has(topic)) {
-            topicSubscriptions.set(topic, new Set());
-        }
-        topicSubscriptions.get(topic).add(socket.id);
-
-        // Add topic to socket
-        if (!socketTopics.has(socket.id)) {
-            socketTopics.set(socket.id, new Set());
-        }
-        socketTopics.get(socket.id).add(topic);
-
-        console.log(`Socket ${socket.id} subscribed to topic: ${topic}`);
-        socket.emit('subscribed', { topic, message: `Successfully subscribed to ${topic}` });
-    });
-
-    // Handle topic unsubscription
-    socket.on('unsubscribe', (topic) => {
-        if (!topic || typeof topic !== 'string') {
-            socket.emit('error', { message: 'Invalid topic' });
-            return;
-        }
-
-        // Remove socket from topic
-        if (topicSubscriptions.has(topic)) {
-            topicSubscriptions.get(topic).delete(socket.id);
-            if (topicSubscriptions.get(topic).size === 0) {
-                topicSubscriptions.delete(topic);
+            return {
+                wsPort,
+                apiPort,
+                baseUrl: `http://127.0.0.1:${wsPort}`,
+            };
+        },
+        async stop() {
+            if (!started) {
+                return;
             }
-        }
 
-        // Remove topic from socket
-        if (socketTopics.has(socket.id)) {
-            socketTopics.get(socket.id).delete(topic);
-            if (socketTopics.get(socket.id).size === 0) {
-                socketTopics.delete(socket.id);
-            }
-        }
+            await new Promise(resolve => io.close(resolve));
 
-        console.log(`Socket ${socket.id} unsubscribed from topic: ${topic}`);
-        socket.emit('unsubscribed', { topic, message: `Successfully unsubscribed from ${topic}` });
-    });
-
-    // Handle disconnect
-    socket.on('disconnect', () => {
-        console.log(`Client disconnected: ${socket.id}`);
-
-        // Clean up subscriptions
-        if (socketTopics.has(socket.id)) {
-            const topics = socketTopics.get(socket.id);
-            topics.forEach(topic => {
-                if (topicSubscriptions.has(topic)) {
-                    topicSubscriptions.get(topic).delete(socket.id);
-                    if (topicSubscriptions.get(topic).size === 0) {
-                        topicSubscriptions.delete(topic);
+            await Promise.all(
+                servers.map(server => new Promise(resolve => {
+                    if (!server.listening) {
+                        return resolve();
                     }
-                }
-            });
-            socketTopics.delete(socket.id);
-        }
-    });
-});
 
-// HTTP API endpoints (Port 8080)
+                    return server.close(resolve);
+                }))
+            );
 
-// Public endpoints (no authentication required)
-app.get('/health', (req, res) => {
-    res.json({
-        status: 'healthy',
-        connections: io.engine.clientsCount,
-        topics: topicSubscriptions.size,
-        timestamp: new Date().toISOString(),
-        apiKeyConfigured: !!API_KEY && API_KEY !== 'default-api-key-change-me'
-    });
-});
+            started = false;
+        },
+    };
+}
 
-app.get('/topics', (req, res) => {
-    const topics = Array.from(topicSubscriptions.keys()).map(topic => ({
-        topic,
-        subscribers: topicSubscriptions.get(topic).size
-    }));
-    res.json({ topics });
-});
+if (require.main === module) {
+    const broker = createBroker(process.env);
 
-app.get('/topics/:topic', (req, res) => {
-    const { topic } = req.params;
-    if (!topicSubscriptions.has(topic)) {
-        return res.status(404).json({ error: 'Topic not found' });
-    }
+    broker
+        .start()
+        .then(({ wsPort, apiPort }) => {
+            console.log(`HTTP API and WebSocket server running on port ${wsPort}`);
+            console.log(`Health check: http://localhost:${wsPort}/health`);
+            console.log(`Publish: POST http://localhost:${wsPort}/publish`);
+            console.log(`API Key configured: ${broker.config.apiKey !== DEFAULT_API_KEY ? 'Yes' : 'No (using default)'}`);
+            console.log(`Auth mode: ${broker.config.authMode}`);
+            console.log(`Token auth configured: ${broker.config.tokenSecret ? 'Yes' : 'No'}`);
 
-    const subscribers = topicSubscriptions.get(topic).size;
-    res.json({ topic, subscribers });
-});
+            if (apiPort !== wsPort) {
+                console.log(`HTTP API running on port ${apiPort}`);
+            }
 
-// Protected endpoints (require API key authentication)
-app.post('/publish', authenticateApiKey, (req, res) => {
-    const { topic, message, data } = req.body;
+            if (broker.config.apiKey === DEFAULT_API_KEY) {
+                console.log('WARNING: Using default API key. Set API_KEY environment variable for production.');
+            }
+        })
+        .catch(error => {
+            console.error(error);
+            process.exit(1);
+        });
 
-    if (!topic || typeof topic !== 'string') {
-        return res.status(400).json({ error: 'Topic is required and must be a string' });
-    }
-
-    if (!message && !data) {
-        return res.status(400).json({ error: 'Either message or data is required' });
-    }
-
-    // Check if topic has subscribers
-    if (!topicSubscriptions.has(topic)) {
-        // Return 204 No Content when there are no subscribers
-        // The operation succeeded, there just weren't any recipients
-        return res.status(204).send();
-    }
-
-    const payload = {
-        topic,
-        message: message || null,
-        data: data || null,
-        timestamp: new Date().toISOString()
+    const shutdown = signal => {
+        console.log(`${signal} received, shutting down gracefully`);
+        broker.stop().then(() => process.exit(0));
     };
 
-    // Emit to all subscribers of the topic
-    const subscribers = topicSubscriptions.get(topic);
-    subscribers.forEach(socketId => {
-        io.to(socketId).emit('message', payload);
-    });
+    process.on('SIGTERM', () => shutdown('SIGTERM'));
+    process.on('SIGINT', () => shutdown('SIGINT'));
+}
 
-    console.log(`Message published to topic '${topic}' for ${subscribers.size} subscribers`);
-
-    res.json({
-        success: true,
-        message: `Message published to topic '${topic}'`,
-        subscribers: subscribers.size,
-        payload
-    });
-});
-
-// Alternative endpoint with topic in URL (also protected)
-app.post('/publish/:topic', authenticateApiKey, (req, res) => {
-    const { topic } = req.params;
-    const { message, data } = req.body;
-
-    if (!message && !data) {
-        return res.status(400).json({ error: 'Either message or data is required' });
-    }
-
-    // Check if topic has subscribers
-    if (!topicSubscriptions.has(topic)) {
-        // Return 204 No Content when there are no subscribers
-        // The operation succeeded, there just weren't any recipients
-        return res.status(204).send();
-    }
-
-    const payload = {
-        topic,
-        message: message || null,
-        data: data || null,
-        timestamp: new Date().toISOString()
-    };
-
-    // Emit to all subscribers of the topic
-    const subscribers = topicSubscriptions.get(topic);
-    subscribers.forEach(socketId => {
-        io.to(socketId).emit('message', payload);
-    });
-
-    console.log(`Message published to topic '${topic}' for ${subscribers.size} subscribers`);
-
-    res.json({
-        success: true,
-        message: `Message published to topic '${topic}'`,
-        subscribers: subscribers.size,
-        payload
-    });
-});
-
-// Start servers
-const WS_PORT = process.env.WS_PORT || 3000;
-const API_PORT = process.env.WS_PORT || 3000; // Use same port for both
-
-// Start HTTP API server (Socket.IO will be served on the same port)
-server.listen(API_PORT, () => {
-    console.log(`HTTP API and WebSocket server running on port ${API_PORT}`);
-    console.log(`Health check: http://localhost:${API_PORT}/health`);
-    console.log(`Topics: http://localhost:${API_PORT}/topics`);
-    console.log(`Publish: POST http://localhost:${API_PORT}/publish`);
-    console.log(`WebSocket connections: ws://localhost:${API_PORT}`);
-    console.log(`API Key configured: ${API_KEY !== 'default-api-key-change-me' ? 'Yes' : 'No (using default)'}`);
-    if (API_KEY === 'default-api-key-change-me') {
-        console.log(`⚠️  WARNING: Using default API key. Set API_KEY environment variable for production!`);
-    }
-});
-
-// Graceful shutdown
-process.on('SIGTERM', () => {
-    console.log('SIGTERM received, shutting down gracefully');
-    server.close(() => {
-        console.log('Process terminated');
-        process.exit(0);
-    });
-});
-
-process.on('SIGINT', () => {
-    console.log('SIGINT received, shutting down gracefully');
-    server.close(() => {
-        console.log('Process terminated');
-        process.exit(0);
-    });
-});
+module.exports = { createBroker };