# MQTT Ingester Stack A Docker-based stack that ingests time-series data from an MQTTS broker into TimescaleDB, exposes it through an authenticated HTTPS API, and optionally forwards logs to Loki. ## Architecture ```mermaid flowchart LR Broker[MQTTS Broker] -->|TLS + client cert| MQTT[MQTT Client] MQTT -->|SQL| TSDB[(TimescaleDB)] MQTT -->|logs| Loki[Loki] TSDB --> API[HTTPS API] Client[API Client] -->|mTLS| API ``` | Service | Purpose | | ------------ | -------------------------------------------------------------- | | `timescaledb`| Time-series database storing telemetry as hypertables. | | `mqtt-client`| Subscribes to MQTTS topics, writes payloads to the database. | | `api` | FastAPI service serving telemetry over HTTPS (mTLS or session).| | `webui` | Web UI for users, groups, and topic access rules. | | `loki` | Optional log aggregation (enabled via the `telemetry` profile).| ## Running Start the core stack: ```bash docker compose up -d ``` Start with Loki enabled: ```bash docker compose --profile telemetry up -d ``` Or use the helper scripts (`scripts/`), which load `.env`, check the certificate files, and print the service URLs: ```bash # PowerShell .\scripts\start.ps1 [-Profiles telemetry] .\scripts\stop.ps1 [-Profiles telemetry] .\scripts\prune.ps1 [-Volumes] [-All] # -Volumes deletes ALL data! # bash ./scripts/start.sh [telemetry] ./scripts/stop.sh [telemetry] ./scripts/prune.sh [--volumes] [--all] ``` The web UI listens on http://localhost:8080 (bound to 127.0.0.1). ## Prerequisites - Docker and Docker Compose v2. - A CA certificate and signed certificates for the MQTT client, the API server, and any API clients. ## Configuration Copy the example environment file and edit it: ```bash cp .env.example .env ``` Key variables: | Variable | Description | | ------------------- | --------------------------------------------- | | `POSTGRES_USER` | TimescaleDB user. | | `POSTGRES_PASSWORD` | TimescaleDB password. | | `MQTT_BROKER_HOST` | MQTTS broker hostname. | | `MQTT_BROKER_PORT` | MQTTS broker port (default `8883`). | | `MQTT_TOPIC` | Topic(s) to subscribe to (default `sensors/#`).| | `MQTT_USERNAME` | MQTT username for broker auth (optional). | | `MQTT_PASSWORD` | MQTT password for broker auth (optional). | | `MQTT_PSK_IDENTITY` | TLS-PSK identity (optional; enables PSK mode).| | `MQTT_PSK_HEX` | TLS-PSK key as hex (optional). | | `DB_BATCH_SIZE` | Rows per batched DB insert (default `500`). | | `DB_FLUSH_SECONDS` | Max seconds a batch waits before flushing (default `1.0`). | | `DB_POOL_MIN_SIZE` | Min API DB pool connections (default `2`). | | `DB_POOL_MAX_SIZE` | Max API DB pool connections (default `10`). | | `LOKI_URL` | Loki push URL; leave empty to disable. | | `LOKI_FLUSH_SECONDS`| Seconds between batched Loki pushes (default `5`). | | `API_PORT` | Host port for the HTTPS API (default `8443`). | | `WEBUI_PORT` | Host port for the web UI (default `8080`). | | `ADMIN_USERNAME` | Initial admin username (default `admin`; created only when no users exist). | | `ADMIN_PASSWORD` | Initial admin password. Empty = default `admin`; the UI then forces a change on first login. | ## Certificates Place the following files in the `certs/` directory: | File | Used by | Purpose | | ------------ | ------------ | -------------------------------- | | `ca.crt` | all | CA certificate. | | `client.crt` | mqtt-client | Client certificate for the broker.| | `client.key` | mqtt-client | Client private key. | | `server.crt` | api | API server certificate. | | `server.key` | api | API server private key. | Generate a self-signed CA and certificates for testing: ```bash mkdir -p certs && cd certs # CA openssl req -x509 -newkey rsa:2048 -days 365 -nodes \ -keyout ca.key -out ca.crt -subj "/CN=Test CA" # MQTT client certificate openssl req -newkey rsa:2048 -nodes -keyout client.key -out client.csr \ -subj "/CN=mqtt-client" openssl x509 -req -in client.csr -CA ca.crt -CAkey ca.key \ -CAcreateserial -out client.crt -days 365 # API server certificate openssl req -newkey rsa:2048 -nodes -keyout server.key -out server.csr \ -subj "/CN=localhost" openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key \ -CAcreateserial -out server.crt -days 365 ``` > The broker must trust `ca.crt` and accept `client.crt` for authentication. ## MQTT Broker Authentication The client always connects over TLS and picks its authentication mode from the environment: | Mode | Variables | Notes | | ---- | --------- | ----- | | **TLS-PSK** | `MQTT_PSK_IDENTITY` + `MQTT_PSK_HEX` | Takes precedence when set. No certificates needed; the broker must have a PSK listener (Mosquitto: `psk_hint` + `psk_file`). Requires Python 3.13+ (the mqtt-client image provides this). | | **Client certificate (mTLS)** | presence of `certs/client.crt` + `certs/client.key` | Default when the files exist and PSK is not configured. | | **Username/password** | `MQTT_USERNAME` (+ `MQTT_PASSWORD`) | Sent in the MQTT CONNECT packet; combinable with either TLS mode. Matches Mosquitto's `password_file`. | | **Anonymous** | none of the above | Only works if the broker allows it; a warning is logged. | Examples: ```bash # TLS-PSK MQTT_PSK_IDENTITY=ingester MQTT_PSK_HEX=0123456789abcdef # Username/password over CA-verified TLS (no client cert required) MQTT_USERNAME=ingester MQTT_PASSWORD=secret ``` TLS-PSK in Mosquitto requires a dedicated listener, e.g.: ``` listener 8884 psk_hint mqtt-ingester psk_file /mosquitto/psk_file # line: ingester:0123456789abcdef ``` ## API The API is served over HTTPS and accepts two authentication methods: - **Mutual TLS**: clients present a certificate signed by the configured CA. Certificate holders are treated as services with full read access. - **Session cookie**: a login in the web UI (http://localhost:8080) issues a `mqtt_session` cookie that also authenticates API requests from the same browser, subject to that user's access rules. ```bash curl --cacert certs/ca.crt \ --cert certs/client.crt \ --key certs/client.key \ https://localhost:8443/health ``` Endpoints: | Method | Path | Description | | ------ | -------------------- | ---------------------------------------- | | GET | `/health` | Health check. | | GET | `/whoami` | Shows the authenticated principal and effective access rules. | | GET | `/telemetry` | List telemetry (filters: `topic`, `device_id`, `limit`). | | GET | `/telemetry/latest` | Latest record (filters: `topic`, `device_id`). | Example: ```bash curl --cacert certs/ca.crt --cert certs/client.crt --key certs/client.key \ "https://localhost:8443/telemetry?topic=sensors/temp&limit=10" ``` ## Web UI and Access Control The web UI (http://localhost:8080) manages **users**, **groups**, and **topic access rules** for the API. Users and groups are stored in TimescaleDB (`db/init/002_access_control.sql`); the API and web UI share the same session store, so one login works for both. ### Users and roles - Every user has a username, an argon2-hashed password, and a role: `user` or `admin`. Admins manage users/groups/rules and implicitly have access to all data; regular users see only what their rules allow. - Users can be disabled (which also kills their sessions) and deleted. - On first start — and only when the users table is empty — an initial admin is created from `ADMIN_USERNAME`/`ADMIN_PASSWORD`, defaulting to `admin`/`admin`. The default password is flagged for a forced change at first login. Restarts never reset or recreate the admin. - Login is throttled: 5 failed attempts per source IP within 15 minutes triggers a 5-minute lockout. ### Access rules A rule is an **allow** or **deny** entry with a glob **pattern** matched against the full tag name (e.g. `11001.KF1.BF1_Value`): | Pattern | Matches | | ------------------- | -------------------------------------------- | | `*` | everything | | `11001.*` | everything prefixed `11001.` | | `11001.*_Value` | `11001.`-prefixed and `_Value`-suffixed tags | | `11001.KF1.*` | everything under the `KF1` function | Semantics: - `*` is the only wildcard and matches any character sequence (including `.`). Everything else — `.`, `_`, digits — is literal. This is deliberately **not** regex: patterns stay writable for plant engineers, translate directly to SQL `LIKE` (so filtering is pushed down into the database and indexes still work), and cannot accidentally match sibling systems the way `^11001.*` would. - A user's effective rules are the **union** of personal rules and the rules of every group they belong to. - **Deny wins over allow**: a tag is readable only if at least one allow rule matches it and no deny rule does. This is what lets a personal deny override a group allow, and vice versa. - Users with no allow rules see nothing (deny-by-default). Rules are enforced in SQL (`topic LIKE ANY(...)` + `NOT (...)`), so `limit` behaves correctly and there is no filtering in application code. Example: a group grants `11001.*`, but a user's personal rule denies `11001.*_Raw` — the user sees the `Value` tags but not the `Raw` ones. ### Sessions Sessions are 12-hour random tokens stored as SHA-256 hashes. The cookie is HttpOnly, SameSite=Lax. All mutating web UI forms require a CSRF token bound to the session. ## Logging The MQTT client logs in JSON format to: - **Console** (always). - **File** (`/var/log/mqtt-client/mqtt-client.log`) when `LOG_TO_FILE=true`. - **Loki** when `LOKI_URL` is set. Records are buffered and pushed in batches from a background thread, so a slow or unavailable Loki never blocks message ingestion. ## Performance - The MQTT client buffers incoming messages in an in-memory queue and inserts them into TimescaleDB in batches (one transaction per batch), keeping database I/O off the MQTT network loop. Tune with `DB_BATCH_SIZE` and `DB_FLUSH_SECONDS`. If the queue fills (broker outpacing the database), the client logs the condition and drops new messages rather than stalling the broker connection. - The API serves queries from a pool of reused database connections (`DB_POOL_MIN_SIZE`/`DB_POOL_MAX_SIZE`). - Chunks older than 7 days are compressed, and data older than 365 days is dropped automatically (see `db/init/001_schema.sql`). ## Database Schema Telemetry is stored in the `telemetry` hypertable: | Column | Type | Description | | ---------- | ----------- | ------------------------------ | | `time` | `TIMESTAMPTZ`| Ingestion timestamp. | | `topic` | `TEXT` | MQTT topic. | | `device_id`| `TEXT` | Extracted from payload if present.| | `payload` | `JSONB` | Full message payload. | Access control (see `db/init/002_access_control.sql`): | Table | Purpose | | -------------- | ---------------------------------------------------------- | | `users` | Accounts: username, argon2 password hash, role, disabled flag. | | `groups` | Named collections of access rules. | | `user_groups` | Group membership. | | `access_rules`| Allow/deny glob patterns owned by a user OR a group. | | `sessions` | Web sessions (SHA-256 token hashes, 12h expiry); shared with the API. | | `login_attempts` | Per-IP login throttling state. | ## Security Notes - Change `POSTGRES_PASSWORD` before deploying. - Set `ADMIN_PASSWORD` in `.env` or change the default `admin`/`admin` immediately after first login (the UI forces it). - Use a real CA and rotate certificates regularly. - Restrict the API port with a firewall or reverse proxy. - The web UI is bound to `127.0.0.1`. If you expose it remotely, put it behind a TLS-terminating reverse proxy; the cookie has no `Secure` flag because the UI itself serves plain HTTP.