diff --git a/Dockers/socket-broker/.gitignore b/Dockers/socket-broker/.gitignore new file mode 100644 index 0000000..2e8157a --- /dev/null +++ b/Dockers/socket-broker/.gitignore @@ -0,0 +1,3 @@ +node_modules/ +.env +*.log diff --git a/Dockers/socket-broker/README.md b/Dockers/socket-broker/README.md index 4191ff2..62df01c 100644 --- a/Dockers/socket-broker/README.md +++ b/Dockers/socket-broker/README.md @@ -1,514 +1,179 @@ # Socket Broker -A WebSocket broker application that handles topic-based messaging between frontend clients and backend services with API key authentication. +A Socket.IO broker with topic-based messaging for ComedyKit. Backend services publish over an HTTP API protected by an API key; browsers and mobile clients authenticate with short-lived, topic-scoped tokens issued by Laravel. ## Features -- **WebSocket Server** (Port 3000): Handles client connections and topic subscriptions -- **HTTP API** (Port 8080): Allows backend services to publish messages to topics -- **API Key Authentication**: Secure access control for publishing messages -- **Topic-based messaging**: Clients subscribe to topics and receive messages when published -- **Real-time communication**: Instant message delivery to all subscribed clients -- **Health monitoring**: Built-in health checks and status endpoints +- Socket.IO server for real-time client subscriptions. +- HTTP API for backend publishing. +- API-key authentication for publishers. +- HMAC-SHA256 client tokens for connection and topic authorization. +- Room-based topic delivery (no in-memory subscription maps). +- Protected topic introspection and public health check. +- CORS allowlist, subscription limits, and subscribe rate limiting. ## Architecture ``` -Frontend Clients ←→ WebSocket Server (Port 3000) - ↕ -Backend Services ←→ HTTP API (Port 8080) [🔐 API Key Required] +Frontend / mobile clients + │ Socket.IO (connection token + topic token) + ▼ +Socket broker ── rooms: user/{uuid}/notifications + ▲ event/{slug}/ticket_statistics + │ POST /publish (API key) +Laravel backend ``` -## Security +## Configuration -**⚠️ IMPORTANT**: The HTTP API endpoints that publish messages require a valid API key for authentication. This ensures that only authorized backend services can send messages to your WebSocket clients. +Copy `env.example` to `.env`: -### API Key Configuration +| Variable | Default | Description | +| --- | --- | --- | +| `WS_PORT` | `3000` | Socket.IO / HTTP port | +| `API_PORT` | `WS_PORT` | Optional second HTTP API port | +| `API_KEY` | `default-api-key-change-me` | Publisher API key; never expose to clients | +| `SOCKET_TOKEN_SECRET` | — | Shared HMAC secret; must match Laravel `SOCKET_TOKEN_SECRET` | +| `AUTH_MODE` | `required` | `off`, `optional` (rollout only), or `required` | +| `ALLOWED_ORIGINS` | `https://*.comedykit.be,...` | Comma-separated origins; supports `*.domain.tld` | +| `MAX_TOPICS_PER_SOCKET` | `20` | Subscription cap per connection | +| `SUBSCRIBE_RATE_LIMIT` | `30` | Subscribe attempts per 10 seconds per connection | -Set the `API_KEY` environment variable to secure your application: +`AUTH_MODE=required` refuses to start without `SOCKET_TOKEN_SECRET`. -```bash -# Set a strong, random API key -export API_KEY="your-secure-api-key-here" +### Rollout -# Or use in docker-compose -API_KEY=your-secure-api-key-here docker-compose up +1. Deploy with `AUTH_MODE=optional` so legacy clients still work. +2. Deploy Laravel with `/notifications/socket-token` and the new publisher job. +3. Deploy the updated web/mobile clients. +4. Switch to `AUTH_MODE=required`. + +## Client authentication + +Laravel issues two token types: + +```json +// Connection token +{ "v": 1, "sub": "", "typ": "connect", "iat": 0, "exp": 0, "jti": "..." } + +// Topic token +{ "v": 1, "sub": "", "typ": "topic", "topic": "event/slug/ticket_statistics", "iat": 0, "exp": 0, "jti": "..." } ``` -### Authentication Methods +Tokens are encoded as: -The API accepts the API key in three ways: - -1. **Authorization Header** (Recommended): - - ```bash - Authorization: Bearer your-api-key-here - ``` - -2. **X-API-Key Header**: - - ```bash - X-API-Key: your-api-key-here - ``` - -3. **Query Parameter**: - ```bash - POST /publish?api_key=your-api-key-here - ``` - -## Quick Start - -### Using Docker - -```bash -# Build the image -docker build -t socket-broker . - -# Run the container with API key -docker run -d \ - --name socket-broker \ - -p 3000:3000 \ - -p 8080:8080 \ - -e API_KEY="your-secure-api-key" \ - socket-broker +``` +base64url(JSON payload) + "." + base64url(HMAC-SHA256(payload, SOCKET_TOKEN_SECRET)) ``` -### Using Docker Compose - -```bash -# Copy environment file -cp env.example .env - -# Edit .env and set your API key -nano .env - -# Start the service -docker-compose up -d -``` - -### Using Node.js directly - -```bash -# Set API key -export API_KEY="your-secure-api-key" - -# Install dependencies -npm install - -# Start the application -npm start -``` - -## Environment Variables - -| Variable | Default | Description | -| ---------- | --------------------------- | ---------------------------------------- | -| `WS_PORT` | `3000` | WebSocket server port | -| `API_PORT` | `8080` | HTTP API server port | -| `API_KEY` | `default-api-key-change-me` | **REQUIRED**: API key for authentication | - -## WebSocket API (Port 3000) - -### Connection - -Connect to `ws://localhost:3000` using Socket.IO client. - -### Events - -#### Subscribe to a topic +Clients connect with the connection token: ```javascript -socket.emit("subscribe", "user-notifications"); -socket.on("subscribed", (data) => { +const socket = io("https://bram-sockets.comedykit.be", { + auth: { token: connectionToken }, +}); +``` + +A user may always subscribe to `user/{their-own-id}/notifications`. Any other topic requires a topic token whose `sub` matches the connection subject and whose `topic` matches the subscription exactly. + +## WebSocket API + +### Subscribe + +```javascript +socket.emit("subscribe", "user//notifications"); + +// Or, for a topic-scoped token: +socket.emit("subscribe", { + topic: "event/my-event/ticket_statistics", + token: topicToken, +}); + +socket.on("subscribed", data => { console.log(`Subscribed to ${data.topic}`); }); ``` -#### Unsubscribe from a topic +### Unsubscribe ```javascript -socket.emit("unsubscribe", "user-notifications"); -socket.on("unsubscribed", (data) => { +socket.emit("unsubscribe", "user//notifications"); +socket.on("unsubscribed", data => { console.log(`Unsubscribed from ${data.topic}`); }); ``` -#### Receive messages +### Receive messages ```javascript -socket.on("message", (data) => { - console.log(`Message from topic ${data.topic}:`, data.message); - console.log("Data:", data.data); - console.log("Timestamp:", data.timestamp); +socket.on("message", data => { + console.log(data.topic, data.message, data.data, data.timestamp); }); ``` -#### Error handling +### Errors ```javascript -socket.on("error", (error) => { - console.error("Socket error:", error.message); +socket.on("error", error => { + // INVALID_TOPIC | FORBIDDEN_TOPIC | RATE_LIMITED | TOPIC_LIMIT + console.error(error.code, error.message); }); ``` -## HTTP API (Port 8080) +## HTTP API -### Public Endpoints (No Authentication Required) +### Public -#### Health Check +- `GET /health` — status, connection count, auth mode. Does not expose topic names. + +### Protected (API key) + +- `GET /topics` +- `GET /topics/:topic` +- `POST /publish` +- `POST /publish/:topic` + +API key is accepted as `Authorization: Bearer `, `x-api-key`, or `api_key` query parameter. + +#### Publish ```bash -GET /health -``` - -Response: - -```json -{ - "status": "healthy", - "connections": 5, - "topics": 3, - "timestamp": "2024-01-01T12:00:00.000Z", - "apiKeyConfigured": true -} -``` - -#### List Topics - -```bash -GET /topics -``` - -Response: - -```json -{ - "topics": [ - { - "topic": "user-notifications", - "subscribers": 2 - }, - { - "topic": "system-alerts", - "subscribers": 1 - } - ] -} -``` - -#### Get Topic Info - -```bash -GET /topics/{topic} -``` - -Response: - -```json -{ - "topic": "user-notifications", - "subscribers": 2 -} -``` - -### Protected Endpoints (API Key Required) - -#### Response Codes - -- **200 OK**: Message published successfully to subscribers -- **204 No Content**: Message published successfully but no subscribers for the topic -- **400 Bad Request**: Invalid request (missing topic, message, or data) -- **401 Unauthorized**: Invalid or missing API key -- **500 Internal Server Error**: Server error - -#### Publish Message - -```bash -POST /publish -Authorization: Bearer your-api-key-here -Content-Type: application/json - -{ - "topic": "user-notifications", - "message": "New message received", - "data": { "userId": 123, "type": "email" } -} -``` - -Alternative endpoint: - -```bash -POST /publish/{topic} -Authorization: Bearer your-api-key-here -Content-Type: application/json - -{ - "message": "New message received", - "data": { "userId": 123, "type": "email" } -} -``` - -Response: - -```json -{ - "success": true, - "message": "Message published to topic 'user-notifications'", - "subscribers": 2, - "payload": { - "topic": "user-notifications", - "message": "New message received", - "data": { "userId": 123, "type": "email" }, - "timestamp": "2024-01-01T12:00:00.000Z" - } -} -``` - -**Note**: If there are no subscribers for the topic, the API returns `204 No Content` instead of an error. This indicates that the message was published successfully, but there were no recipients to receive it. - -### Authentication Errors - -If you don't provide a valid API key, you'll get: - -```json -{ - "error": "Unauthorized", - "message": "API key is required. Use Authorization header or x-api-key header or api_key query parameter." -} -``` - -## Frontend Example (JavaScript) - -```html - - - - Socket Broker Client - - - -

Socket Broker Client

-
Connecting...
-
- - - - -``` - -## Backend Example (Python) - -```python -import requests -import json -import os - -# Get API key from environment variable -API_KEY = os.getenv('API_KEY', 'your-api-key-here') - -def publish_message(topic, message, data=None): - url = "http://localhost:8080/publish" - headers = { - 'Authorization': f'Bearer {API_KEY}', - 'Content-Type': 'application/json' - } - - payload = { - "topic": topic, - "message": message, - "data": data - } - - response = requests.post(url, json=payload, headers=headers) - return response.json() - -# Example usage -result = publish_message( - topic="user-notifications", - message="New email received", - data={"userId": 123, "emailId": "abc123"} -) -print(result) -``` - -## Backend Example (Node.js) - -```javascript -const axios = require("axios"); - -const API_KEY = process.env.API_KEY || "your-api-key-here"; - -async function publishMessage(topic, message, data = null) { - try { - const response = await axios.post( - "http://localhost:8080/publish", - { - topic, - message, - data, - }, - { - headers: { - Authorization: `Bearer ${API_KEY}`, - "Content-Type": "application/json", - }, - } - ); - - return response.data; - } catch (error) { - console.error( - "Error publishing message:", - error.response?.data || error.message - ); - throw error; - } -} - -// Example usage -publishMessage("user-notifications", "New notification", { userId: 123 }) - .then((result) => console.log("Message published:", result)) - .catch((error) => console.error("Failed to publish:", error)); -``` - -## Docker Compose Example - -```yaml -version: "3.8" -services: - socket-broker: - build: . - ports: - - "3000:3000" - - "8080:8080" - environment: - - NODE_ENV=production - - WS_PORT=3000 - - API_PORT=8080 - - API_KEY=${API_KEY:-default-api-key-change-me} - restart: unless-stopped - healthcheck: - test: - [ - "CMD", - "node", - "-e", - "require('http').get('http://localhost:8080/health', (res) => { process.exit(res.statusCode === 200 ? 0 : 1) })", - ] - interval: 30s - timeout: 10s - retries: 3 - start_period: 40s - env_file: - - .env -``` - -## Monitoring - -The application provides several monitoring endpoints: - -- `/health` - Overall health status and API key configuration status -- `/topics` - List of active topics and subscriber counts -- `/topics/{topic}` - Specific topic information - -## Security Best Practices - -1. **Generate Strong API Keys**: Use cryptographically secure random strings - - ```bash - # Generate a secure API key - openssl rand -base64 32 - ``` - -2. **Environment Variables**: Never hardcode API keys in your code - - ```bash - # Good - export API_KEY="$(openssl rand -base64 32)" - - # Bad - export API_KEY="my-secret-key-123" - ``` - -3. **Rotate Keys Regularly**: Change your API keys periodically - -4. **Limit Access**: Only share API keys with trusted backend services - -5. **HTTPS in Production**: Use HTTPS/WSS in production environments - -6. **Network Security**: Consider using internal networks for backend communication - -## Troubleshooting - -### Common Issues - -1. **401 Unauthorized**: Check that you're providing the correct API key -2. **Port conflicts**: Ensure ports 3000 and 8080 are available -3. **CORS errors**: Check that the frontend origin is allowed -4. **Connection refused**: Verify the container is running and ports are exposed - -### Testing Authentication - -Test your API key authentication: - -```bash -# Test without API key (should fail with 401) -curl -X POST http://localhost:8080/publish \ +curl -X POST http://localhost:3000/publish \ + -H "x-api-key: your-api-key" \ -H "Content-Type: application/json" \ - -d '{"topic":"test","message":"hello"}' - -# Test with correct API key but no subscribers (should succeed with 204) -curl -X POST http://localhost:8080/publish \ - -H "Authorization: Bearer your-api-key" \ - -H "Content-Type: application/json" \ - -d '{"topic":"test","message":"hello"}' - -# Test with correct API key and subscribers (should succeed with 200) -curl -X POST http://localhost:8080/publish \ - -H "Authorization: Bearer your-api-key" \ - -H "Content-Type: application/json" \ - -d '{"topic":"test","message":"hello"}' + -d '{"topic":"user//notifications","message":"{\"id\":\"...\"}"}' ``` -### Logs +Response: -Check application logs for detailed information: +- `200` with `{ success, message, subscribers, payload }` +- `204` when the topic has no subscribers +- `400` invalid request +- `401` invalid or missing API key + +## Development ```bash -docker logs socket-broker +npm install +npm test +AUTH_MODE=optional API_KEY=dev-key SOCKET_TOKEN_SECRET=dev-secret npm start ``` -### Health Check - -Verify the application is healthy and API key is configured: +## Docker ```bash -curl http://localhost:8080/health +cp env.example .env +# set API_KEY, SOCKET_TOKEN_SECRET, AUTH_MODE, ALLOWED_ORIGINS +docker compose up -d ``` + +The Docker image runs as a non-root user and exposes port 3000. If `API_PORT` differs from `WS_PORT`, a second HTTP server is started on that port. + +## Security notes + +- `API_KEY` is publisher-only. Never ship it to clients. +- `SOCKET_TOKEN_SECRET` must be at least 32 random bytes and identical in Laravel and the broker. +- Tokens are short-lived. The broker verifies them when a subscription is created; clients re-fetch tokens on reconnect. +- Use HTTPS/WSS in production. +- Rotate secrets by coordinating Laravel and broker deployments; outstanding tokens become invalid. +- `AUTH_MODE=optional` or `off` is for local development or the rollout window only. diff --git a/Dockers/socket-broker/docker-compose.yml b/Dockers/socket-broker/docker-compose.yml index fcee2be..bcd8a15 100644 --- a/Dockers/socket-broker/docker-compose.yml +++ b/Dockers/socket-broker/docker-compose.yml @@ -12,6 +12,11 @@ services: - WS_PORT=3000 - API_PORT=8080 - API_KEY=${API_KEY:-default-api-key-change-me} + - SOCKET_TOKEN_SECRET=${SOCKET_TOKEN_SECRET} + - AUTH_MODE=${AUTH_MODE:-required} + - ALLOWED_ORIGINS=${ALLOWED_ORIGINS:-https://*.comedykit.be,https://comedykit.be,https://www.comedykit.be,http://localhost:5173,http://localhost:3000,capacitor://localhost,http://localhost,https://localhost} + - MAX_TOPICS_PER_SOCKET=${MAX_TOPICS_PER_SOCKET:-20} + - SUBSCRIBE_RATE_LIMIT=${SUBSCRIBE_RATE_LIMIT:-30} restart: unless-stopped healthcheck: test: diff --git a/Dockers/socket-broker/env.example b/Dockers/socket-broker/env.example index 3f19283..8b5ac36 100644 --- a/Dockers/socket-broker/env.example +++ b/Dockers/socket-broker/env.example @@ -4,13 +4,29 @@ # WebSocket server port WS_PORT=3000 -# HTTP API server port +# HTTP API port (set equal to WS_PORT to serve both on one port) API_PORT=8080 -# API Key for authentication (REQUIRED for production) -# This key must be provided in HTTP requests to publish messages -# Generate a strong, random key for production use +# API Key for publishing messages (REQUIRED for production) +# This key is only for backend publishers; do not expose it to clients. API_KEY=your-secure-api-key-here +# Shared HMAC secret used to verify short-lived client subscription tokens. +# Must match SOCKET_TOKEN_SECRET in the Laravel backend. +SOCKET_TOKEN_SECRET=your-socket-token-secret-here + +# Client authentication mode: +# off - no client auth (local development only) +# optional - accept valid tokens, allow legacy unauthenticated clients (rollout only) +# required - reject unauthenticated clients (production) +AUTH_MODE=required + +# Comma-separated allowed origins. Supports exact origins and *.domain.tld. +ALLOWED_ORIGINS=https://*.comedykit.be,https://comedykit.be,https://www.comedykit.be,http://localhost:5173,http://localhost:3000,capacitor://localhost,http://localhost,https://localhost + +# Subscription limits +MAX_TOPICS_PER_SOCKET=20 +SUBSCRIBE_RATE_LIMIT=30 + # Node environment -NODE_ENV=production \ No newline at end of file +NODE_ENV=production diff --git a/Dockers/socket-broker/index.js b/Dockers/socket-broker/index.js index 0cacd75..0d27da1 100644 --- a/Dockers/socket-broker/index.js +++ b/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 }; diff --git a/Dockers/socket-broker/package-lock.json b/Dockers/socket-broker/package-lock.json index aa0e0f9..0a25785 100644 --- a/Dockers/socket-broker/package-lock.json +++ b/Dockers/socket-broker/package-lock.json @@ -14,7 +14,8 @@ "socket.io": "^4.7.2" }, "devDependencies": { - "nodemon": "^3.0.1" + "nodemon": "^3.0.1", + "socket.io-client": "^4.7.2" } }, "node_modules/@socket.io/component-emitter": { @@ -347,6 +348,63 @@ "node": ">=10.2.0" } }, + "node_modules/engine.io-client": { + "version": "6.6.7", + "resolved": "https://registry.npmjs.org/engine.io-client/-/engine.io-client-6.6.7.tgz", + "integrity": "sha512-GyAvmPA7gnRCpDQbYACAy79Lg4ewevrmaPzySZz0OVvgyZc3gEfLLqPTYk8wVBScjOEpM9ZO4KryyFpVItjGEA==", + "dev": true, + "dependencies": { + "@socket.io/component-emitter": "~3.1.0", + "debug": "~4.4.1", + "engine.io-parser": "~5.2.1", + "ws": "~8.21.0", + "xmlhttprequest-ssl": "~2.1.1" + } + }, + "node_modules/engine.io-client/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "dev": true, + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/engine.io-client/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "dev": true + }, + "node_modules/engine.io-client/node_modules/ws": { + "version": "8.21.3", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.3.tgz", + "integrity": "sha512-201TZ/kPWxoPr/OKWjquZR1SWKXcvxdH+e1xrx89b3YbmzLMFCLfnaG1HFIgWzJOEWZ7MvpK++odZufgYR50Rw==", + "dev": true, + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + }, "node_modules/engine.io-parser": { "version": "5.2.3", "resolved": "https://registry.npmjs.org/engine.io-parser/-/engine.io-parser-5.2.3.tgz", @@ -1273,6 +1331,44 @@ "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", "license": "MIT" }, + "node_modules/socket.io-client": { + "version": "4.8.4", + "resolved": "https://registry.npmjs.org/socket.io-client/-/socket.io-client-4.8.4.tgz", + "integrity": "sha512-7wjKAvVXYOJsbmUgkWbnhPD4bFpkqrUR0++XOjpGFmqDqsqV1CI93VTraeExLxpjxEqrWgFoSRZu1ihjLWfkAA==", + "dev": true, + "dependencies": { + "@socket.io/component-emitter": "~3.1.0", + "debug": "~4.4.1", + "engine.io-client": "~6.6.1", + "socket.io-parser": "~4.2.4" + }, + "engines": { + "node": ">=10.0.0" + } + }, + "node_modules/socket.io-client/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "dev": true, + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/socket.io-client/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "dev": true + }, "node_modules/socket.io-parser": { "version": "4.2.4", "resolved": "https://registry.npmjs.org/socket.io-parser/-/socket.io-parser-4.2.4.tgz", @@ -1459,6 +1555,15 @@ "optional": true } } + }, + "node_modules/xmlhttprequest-ssl": { + "version": "2.1.2", + "resolved": "https://registry.npmjs.org/xmlhttprequest-ssl/-/xmlhttprequest-ssl-2.1.2.tgz", + "integrity": "sha512-TEU+nJVUUnA4CYJFLvK5X9AOeH4KvDvhIfm0vV1GaQRtchnG0hgK5p8hw/xjv8cunWYCsiPCSDzObPyhEwq3KQ==", + "dev": true, + "engines": { + "node": ">=0.4.0" + } } } } diff --git a/Dockers/socket-broker/package.json b/Dockers/socket-broker/package.json index bdbd5e1..552536d 100644 --- a/Dockers/socket-broker/package.json +++ b/Dockers/socket-broker/package.json @@ -5,7 +5,8 @@ "main": "index.js", "scripts": { "start": "node index.js", - "dev": "nodemon index.js" + "dev": "nodemon index.js", + "test": "node --test test/*.test.js" }, "dependencies": { "express": "^4.18.2", @@ -13,7 +14,8 @@ "cors": "^2.8.5" }, "devDependencies": { - "nodemon": "^3.0.1" + "nodemon": "^3.0.1", + "socket.io-client": "^4.7.2" }, "keywords": [ "websocket", diff --git a/Dockers/socket-broker/test/socket-broker.test.js b/Dockers/socket-broker/test/socket-broker.test.js new file mode 100644 index 0000000..0e4dc71 --- /dev/null +++ b/Dockers/socket-broker/test/socket-broker.test.js @@ -0,0 +1,267 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const crypto = require('node:crypto'); +const http = require('node:http'); +const { io } = require('socket.io-client'); +const { createBroker } = require('../index'); + +const API_KEY = 'broker-test-api-key'; +const TOKEN_SECRET = 'broker-test-token-secret'; +const MAX_SKEW_SECONDS = 60; + +const sockets = new Set(); +let baseUrl = null; + +const broker = createBroker({ + NODE_ENV: 'test', + WS_PORT: '0', + API_PORT: '0', + API_KEY, + SOCKET_TOKEN_SECRET: TOKEN_SECRET, + AUTH_MODE: 'required', + ALLOWED_ORIGINS: '*', +}); + +const startPromise = broker.start().then(info => { + baseUrl = info.baseUrl; + return info; +}); + +function makeToken(payload) { + const encodedPayload = Buffer.from(JSON.stringify(payload)).toString('base64url'); + const signature = crypto + .createHmac('sha256', TOKEN_SECRET) + .update(encodedPayload) + .digest('base64url'); + + return `${encodedPayload}.${signature}`; +} + +function connectToken(subject) { + return makeToken({ + v: 1, + sub: subject, + typ: 'connect', + iat: Math.floor(Date.now() / 1000), + exp: Math.floor(Date.now() / 1000) + 600, + }); +} + +function topicToken(subject, topic) { + return makeToken({ + v: 1, + sub: subject, + typ: 'topic', + topic, + iat: Math.floor(Date.now() / 1000), + exp: Math.floor(Date.now() / 1000) + 600, + }); +} + +function randomUuid() { + return crypto.randomUUID(); +} + +function connect(auth = {}) { + const socket = io(baseUrl, { + auth, + transports: ['websocket'], + reconnection: false, + forceNew: true, + timeout: 2000, + }); + + sockets.add(socket); + + return socket; +} + +function once(socket, event, timeoutMs = 3000) { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(`Timed out waiting for ${event}`)), timeoutMs); + + socket.once(event, payload => { + clearTimeout(timer); + resolve(payload); + }); + }); +} + +function closeAll() { + for (const socket of sockets) { + socket.close(); + } + + sockets.clear(); +} + +function request(method, requestPath, options = {}) { + return new Promise((resolve, reject) => { + const body = options.body ? JSON.stringify(options.body) : null; + const headers = { ...(options.headers || {}) }; + + if (body) { + headers['content-type'] = 'application/json'; + headers['content-length'] = Buffer.byteLength(body); + } + + const req = http.request(`${baseUrl}${requestPath}`, { method, headers }, response => { + let raw = ''; + response.setEncoding('utf8'); + response.on('data', chunk => { + raw += chunk; + }); + response.on('end', () => { + let parsed = raw; + + try { + parsed = raw ? JSON.parse(raw) : null; + } catch { + // Keep the raw response for assertions. + } + + resolve({ status: response.statusCode, body: parsed, raw }); + }); + }); + + req.on('error', reject); + + if (body) { + req.write(body); + } + + req.end(); + }); +} + +test('socket broker', async t => { + try { + await startPromise; + + await t.test('health is public and does not expose topic names', async () => { + const response = await request('GET', '/health'); + + assert.equal(response.status, 200); + assert.equal(response.body.status, 'healthy'); + assert.equal(response.body.authMode, 'required'); + assert.equal(response.body.tokenAuthConfigured, true); + assert.equal(response.body.topics, undefined); + }); + + await t.test('topic introspection requires the API key', async () => { + const unauthorized = await request('GET', '/topics'); + assert.equal(unauthorized.status, 401); + + const authorized = await request('GET', '/topics', { + headers: { 'x-api-key': API_KEY }, + }); + assert.equal(authorized.status, 200); + }); + + await t.test('connections without a token are rejected', async () => { + const socket = connect({}); + const error = await once(socket, 'connect_error'); + + assert.match(error.message, /unauthorized/i); + socket.close(); + }); + + await t.test('tampered and expired tokens are rejected', async () => { + const tampered = connect({ token: `${connectToken(randomUuid())}tampered` }); + const tamperedError = await once(tampered, 'connect_error'); + assert.match(tamperedError.message, /unauthorized/i); + tampered.close(); + + const expiredToken = makeToken({ + v: 1, + sub: randomUuid(), + typ: 'connect', + iat: Math.floor(Date.now() / 1000) - 3600, + exp: Math.floor(Date.now() / 1000) - MAX_SKEW_SECONDS - 60, + }); + const expired = connect({ token: expiredToken }); + const expiredError = await once(expired, 'connect_error'); + assert.match(expiredError.message, /unauthorized/i); + expired.close(); + }); + + await t.test('a user can subscribe to their own notification topic only', async () => { + const userId = randomUuid(); + const otherUserId = randomUuid(); + const socket = connect({ token: connectToken(userId) }); + + await once(socket, 'connect'); + + socket.emit('subscribe', `user/${userId}/notifications`); + const subscribed = await once(socket, 'subscribed'); + assert.equal(subscribed.topic, `user/${userId}/notifications`); + + socket.emit('subscribe', `user/${otherUserId}/notifications`); + const error = await once(socket, 'error'); + assert.equal(error.code, 'FORBIDDEN_TOPIC'); + + socket.close(); + }); + + await t.test('event stats require a matching topic token', async () => { + const userId = randomUuid(); + const eventTopic = 'event/test-event/ticket_statistics'; + const socket = connect({ token: connectToken(userId) }); + + await once(socket, 'connect'); + + socket.emit('subscribe', { + topic: eventTopic, + token: topicToken(userId, eventTopic), + }); + const subscribed = await once(socket, 'subscribed'); + assert.equal(subscribed.topic, eventTopic); + + const otherTopic = 'event/other-event/ticket_statistics'; + socket.emit('subscribe', { + topic: otherTopic, + token: topicToken(userId, eventTopic), + }); + const error = await once(socket, 'error'); + assert.equal(error.code, 'FORBIDDEN_TOPIC'); + + socket.close(); + }); + + await t.test('publish delivers to subscribers and unsubscribe removes the room', async () => { + const userId = randomUuid(); + const topic = `user/${userId}/notifications`; + const socket = connect({ token: connectToken(userId) }); + + await once(socket, 'connect'); + socket.emit('subscribe', topic); + await once(socket, 'subscribed'); + + const messagePromise = once(socket, 'message'); + const publishResponse = await request('POST', '/publish', { + headers: { 'x-api-key': API_KEY }, + body: { topic, message: '{"id":"notification-1"}' }, + }); + + assert.equal(publishResponse.status, 200); + + const message = await messagePromise; + assert.equal(message.topic, topic); + assert.equal(message.message, '{"id":"notification-1"}'); + + socket.emit('unsubscribe', topic); + await once(socket, 'unsubscribed'); + + const emptyResponse = await request('POST', '/publish', { + headers: { 'x-api-key': API_KEY }, + body: { topic, message: '{"id":"notification-2"}' }, + }); + + assert.equal(emptyResponse.status, 204); + socket.close(); + }); + } finally { + closeAll(); + await broker.stop(); + } +}); diff --git a/Dockers/socket-broker/test_api.py b/Dockers/socket-broker/test_api.py index 1acbb31..3e7975e 100644 --- a/Dockers/socket-broker/test_api.py +++ b/Dockers/socket-broker/test_api.py @@ -33,8 +33,9 @@ def test_health(): print(f"✅ Health check passed") print(f" Status: {data['status']}") print(f" Connections: {data['connections']}") - print(f" Topics: {data['topics']}") + print(f" Topics: {data.get('topics', 'n/a')}") print(f" Timestamp: {data['timestamp']}") + print(f" Auth Mode: {data.get('authMode', 'Unknown')}") print(f" API Key Configured: {data.get('apiKeyConfigured', 'Unknown')}") return True else: @@ -45,10 +46,10 @@ def test_health(): return False def test_topics(): - """Test the topics endpoint""" + """Test the topics endpoint (requires API key)""" print("\nTesting topics endpoint...") try: - response = requests.get(f"{API_BASE_URL}/topics") + response = requests.get(f"{API_BASE_URL}/topics", headers=get_auth_headers()) if response.status_code == 200: data = response.json() print(f"✅ Topics retrieved successfully") @@ -66,10 +67,10 @@ def test_topics(): return False def test_topic_info(topic): - """Test getting info for a specific topic""" + """Test getting info for a specific topic (requires API key)""" print(f"\nTesting topic info for '{topic}'...") try: - response = requests.get(f"{API_BASE_URL}/topics/{topic}") + response = requests.get(f"{API_BASE_URL}/topics/{topic}", headers=get_auth_headers()) if response.status_code == 200: data = response.json() print(f"✅ Topic info retrieved successfully")