mqtt-ingester/README.md

# 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.