511 lines
15 KiB
JavaScript
511 lines
15 KiB
JavaScript
const crypto = require('crypto');
|
|
const express = require('express');
|
|
const { createServer } = require('http');
|
|
const { Server } = require('socket.io');
|
|
const cors = require('cors');
|
|
|
|
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',
|
|
];
|
|
|
|
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.`);
|
|
}
|
|
|
|
const tokenSecret = env.SOCKET_TOKEN_SECRET || '';
|
|
|
|
if (authMode === 'required' && !tokenSecret) {
|
|
throw new Error('SOCKET_TOKEN_SECRET is required when AUTH_MODE=required.');
|
|
}
|
|
|
|
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();
|
|
}
|
|
|
|
return res.status(401).json({
|
|
error: 'Unauthorized',
|
|
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);
|
|
});
|
|
}
|
|
|
|
return {
|
|
config,
|
|
app,
|
|
io,
|
|
async start() {
|
|
const wsPort = await listen(wsServer, config.wsPort);
|
|
const apiPort = apiServer ? await listen(apiServer, config.apiPort) : wsPort;
|
|
started = true;
|
|
|
|
return {
|
|
wsPort,
|
|
apiPort,
|
|
baseUrl: `http://127.0.0.1:${wsPort}`,
|
|
};
|
|
},
|
|
async stop() {
|
|
if (!started) {
|
|
return;
|
|
}
|
|
|
|
await new Promise(resolve => io.close(resolve));
|
|
|
|
await Promise.all(
|
|
servers.map(server => new Promise(resolve => {
|
|
if (!server.listening) {
|
|
return resolve();
|
|
}
|
|
|
|
return server.close(resolve);
|
|
}))
|
|
);
|
|
|
|
started = false;
|
|
},
|
|
};
|
|
}
|
|
|
|
if (require.main === module) {
|
|
const broker = createBroker(process.env);
|
|
|
|
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'}`);
|
|
|
|
if (apiPort !== wsPort) {
|
|
console.log(`HTTP API running on port ${apiPort}`);
|
|
}
|
|
|
|
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);
|
|
});
|
|
|
|
const shutdown = signal => {
|
|
console.log(`${signal} received, shutting down gracefully`);
|
|
broker.stop().then(() => process.exit(0));
|
|
};
|
|
|
|
process.on('SIGTERM', () => shutdown('SIGTERM'));
|
|
process.on('SIGINT', () => shutdown('SIGINT'));
|
|
}
|
|
|
|
module.exports = { createBroker };
|