diff --git a/.env.dev b/.env.dev new file mode 100644 index 0000000..b96e8c8 --- /dev/null +++ b/.env.dev @@ -0,0 +1,64 @@ +# --------------------------------------------------------------------------- +# Copy to .env (docker compose) or paste into Portainer's "Environment +# variables" section when deploying the stack. Everything has a working +# default except POSTGRES_PASSWORD. +# --------------------------------------------------------------------------- + +# PostgreSQL (db container). Pick a strong password; avoid characters that are +# reserved in URLs (@ : / ? # % &) or URL-encode them, since the password is +# embedded in the DB_DSN connection string. +POSTGRES_USER=nodeledge +POSTGRES_PASSWORD=u9UHTgucnH7nZ5S9Z38ilDpiasE17rWc +POSTGRES_DB=nodeledge + +# MQTT broker the listener subscribes to. The default points at the optional +# "broker" test service; in production set this to your real broker host. +MQTT_BROKER_HOST=oem.nodeledge.se +MQTT_BROKER_PORT=1883 +MQTT_USERNAME= +MQTT_PASSWORD= + +# Subscription filter. Default "/+/+" matches topics on the form +# //. Use "#", "/up/#", ... to widen it (the IMEI is then +# always taken from the LAST topic segment). +MQTT_TOPIC_FILTER=vivab/# +# 0 = at most once, 1 = at least once (duplicate rows possible after reconnects) +MQTT_QOS=1 +# Unique per deployment! Two clients with the same id on one broker kick each +# other off in a reconnect loop (the local dev instance and the production +# stack must not share this value). +MQTT_CLIENT_ID=nodeledge-mqtt-listener-local-test + +# Keys the listener injects into every payload. +INJECT_IMEI_KEY=imei +INJECT_DEVICE_KEY=device + +# API authentication bootstrap. On the FIRST deployment the app_tokens table +# is empty; on the first authenticated request the api seeds the SHA-256 +# digest of this token. Set a long random value, rotate to real per-app +# tokens via PUT /configs/app_tokens, then unset this. Ignored once the +# table has rows (and never on a migrated database). +BOOTSTRAP_APP_TOKEN=IPyoPDwPNMOeHEd9QsBM42NrLvJTZx3Qf4CKCJakDrlJ4Qwz0BwKdRMB56AAIIGb +# Label for the seeded bootstrap row (informational only). +BOOTSTRAP_APP_NAME=bootstrap + +# Comma-separated payload keys searched for a device timestamp (first +# parseable one wins). Accepts ISO 8601 strings, epoch seconds or epoch +# milliseconds. Rows without a parseable timestamp keep payload_time NULL. +PAYLOAD_TIME_KEYS=timestamp,time,ts + +# Host ports +API_HOST_PORT=18080 +MQTT_HOST_PORT=2883 +PGADMIN_HOST_PORT=8092 + +# API: hard cap for the limit query parameter. +API_MAX_LIMIT=10000 + +LOG_LEVEL=INFO + + +PGADMIN_DEFAULT_EMAIL=jonas@bengtsson.ws +PGADMIN_DEFAULT_PASSWORD=change_this_bad_password +PGADMIN_DISABLE_POSTFIX=true +PGADMIN_LISTEN_ADDRESS=0.0.0.0 diff --git a/.env.example b/.env.example index 407adec..65780d3 100644 --- a/.env.example +++ b/.env.example @@ -32,6 +32,15 @@ MQTT_CLIENT_ID=nodeledge-mqtt-listener INJECT_IMEI_KEY=imei INJECT_DEVICE_KEY=device +# API authentication bootstrap. On the FIRST deployment the app_tokens table +# is empty; on the first authenticated request the api seeds the SHA-256 +# digest of this token. Set a long random value, rotate to real per-app +# tokens via PUT /configs/app_tokens, then unset this. Ignored once the +# table has rows (and never on a migrated database). +BOOTSTRAP_APP_TOKEN= +# Label for the seeded bootstrap row (informational only). +BOOTSTRAP_APP_NAME=bootstrap + # Comma-separated payload keys searched for a device timestamp (first # parseable one wins). Accepts ISO 8601 strings, epoch seconds or epoch # milliseconds. Rows without a parseable timestamp keep payload_time NULL. diff --git a/README.md b/README.md index 0a58763..ada846d 100644 --- a/README.md +++ b/README.md @@ -3,8 +3,8 @@ A Portainer / docker compose stack that: 1. **mqtt-listener** — subscribes to MQTT topics on the form `//`, injects `imei` and a `device` name into each JSON object of the payload (a single object or an array of objects), and stores every object as its own row in PostgreSQL. -2. **api** — a FastAPI read API over the database, protected by static app tokens. -3. **db** — PostgreSQL 16 storing the messages (JSONB payloads). +2. **api** — a FastAPI read API over the database, protected by app tokens stored in the database as SHA-256 digests. +3. **db** — PostgreSQL 16 storing the messages (JSONB payloads) and the config tables (`app_tokens`, `imei_map`). ``` ┌─────────────┐ /xxx/IMEI ┌───────────────┐ INSERT ┌────┐ @@ -21,9 +21,7 @@ A Portainer / docker compose stack that: ``` docker-compose.yml stack definition (paste into Portainer) .env.example environment variables (POSTGRES_PASSWORD etc.) -init/01_schema.sql creates the mqtt_messages table on first db start -configs/listener/imei_map.json IMEI -> device name mapping -configs/api/app_tokens.json app tokens for the API +init/01_schema.sql creates mqtt_messages + app_tokens + imei_map on first db start listener/ mqtt -> db service (paho-mqtt, psycopg) api/ FastAPI service (uvicorn) mosquitto/ config for the optional test broker (profile "test") @@ -37,18 +35,18 @@ cp .env.example .env # then edit POSTGRES_PASSWORD docker compose --profile test up -d --build ``` -This starts db, mqtt-listener, api and an anonymous Mosquitto broker (ports 1883 → host 2883, API 18080 → host 18080). Publish some messages: +This starts db, mqtt-listener, api and an anonymous Mosquitto broker (ports 1883 → host 2883, API 18080 → host 18080). Set `BOOTSTRAP_APP_TOKEN` to a long random value in `.env` first — it seeds the initially empty `app_tokens` table on the first authenticated request. Then publish some messages: ```bash pip install -r tools/requirements.txt python tools/publish_test_messages.py --broker localhost:1883 ``` -Fetch data from the API: +Fetch data from the API (use the bootstrap token you set): ```bash python tools/smoke_test_api.py --url http://localhost:18080 \ - --token replace-me-with-a-long-random-secret-1 + --token "$BOOTSTRAP_APP_TOKEN" ``` Expected smoke test output: @@ -68,11 +66,11 @@ PASS: count endpoint returned {'count': N} - `MQTT_BROKER_HOST` — your real broker hostname/IP (not the test broker). - `MQTT_USERNAME` / `MQTT_PASSWORD` — broker credentials, if required. - Optional: `MQTT_TOPIC_FILTER`, `MQTT_QOS`, `API_MAX_LIMIT`, `LOG_LEVEL`, ports. -3. Because the stack references local files (`init/01_schema.sql`, `configs/...`, Dockerfiles), the stack files must be present on the Docker host. Two options: +3. Because the stack references local files (`init/01_schema.sql`, Dockerfiles), the stack files must be present on the Docker host. Two options: - **Recommended:** put the repository in a folder on the host (e.g. `/opt/nodeledge`) and use **Stacks → Add stack → Repository** pointing at your git repo, or - Upload the files via "Custom template" including all files. -4. **Secrets:** edit `configs/api/app_tokens.json` and `configs/listener/imei_map.json` on the host to hold your real values before deploying (the defaults are placeholders). -5. Deploy the stack. The `db` service runs `init/01_schema.sql` on its first start (an empty volume), creating the `mqtt_messages` table. +4. **Tokens:** set `BOOTSTRAP_APP_TOKEN` (a long random secret) in the stack's environment. On the first authenticated request after a fresh deploy, the api seeds its SHA-256 digest into the empty `app_tokens` table. Rotate to real per-app tokens via `PUT /configs/app_tokens`, then remove the bootstrap variable (it is ignored once the table has rows). +5. Deploy the stack. The `db` service runs `init/01_schema.sql` on its first start (an empty volume), creating `mqtt_messages`, `app_tokens` and `imei_map`. 6. Verify: Stacks → your stack → check that db, mqtt-listener and api are all healthy (green). > The `broker` service only runs when the compose profile `test` is active (Portainer: enable the `test` profile in the stack settings), so production deploys never spin up an extra broker. @@ -81,7 +79,7 @@ PASS: count endpoint returned {'count': N} - The IMEI is the **last segment** of the topic, e.g. `/vivab/350123456789012` → `350123456789012`. - It must be exactly **15 digits**; anything else (letters, separators, other lengths) is skipped. -- It looks the IMEI up in `configs/listener/imei_map.json`; on a match the device name is injected, otherwise the IMEI itself is used as the device name. +- It looks the IMEI up in the `imei_map` table; on a match the device name is injected, otherwise the IMEI itself is used as the device name. The listener re-reads the table every 5 seconds, so API edits take effect without a restart. - The injected keys are `imei` and `device` (configurable via `INJECT_IMEI_KEY` / `INJECT_DEVICE_KEY`). Any existing values in the payload are **overwritten** — the topic is authoritative. - A **device timestamp** is extracted at ingest time from the first parseable payload key in `PAYLOAD_TIME_KEYS` (default `timestamp,time,ts`; ISO 8601 strings, epoch seconds and epoch milliseconds all accepted) and stored in the `payload_time` column. A missing or unparseable timestamp does **not** skip the message — `payload_time` is simply NULL and `received_at` (server receipt time) remains authoritative. - The payload can be a **single JSON object or a JSON array of objects**. Each object in an array is injected and stored as its own database row (each with its own `payload_time`); array elements that are not objects are skipped. @@ -96,33 +94,47 @@ Authentication: pass one of the configured app tokens either as the `X-App-Token | `GET /health` | none | Liveness probe (public) | | `GET /messages/count` | token | Total number of stored messages | | `GET /messages` | token | List messages, newest first. Query parameters: `imei`, `device` (exact-match filters), `from`/`to` (ISO 8601 range on `received_at` — server receipt time; inclusive; naive values treated as UTC), `payload_from`/`payload_to` (same but on `payload_time` — the device timestamp extracted from the JSON; messages without one are excluded; ordering switches to `payload_time DESC`), `limit` (default 100, capped at `API_MAX_LIMIT`), `offset` for paging. Reversed ranges return 422 | -| `GET /configs/{name}` | token | Read a config file's content as JSON. `name`: `app_tokens` or `imei_map` | -| `PUT/POST /configs/{name}` | token | Replace a config file's content. Body **must be a JSON object**; per-config schema is validated (imei_map keys must be 15-digit IMEIs; app_tokens must keep the `{"tokens": {name: secret}}` shape). Invalid payloads → 422, unknown names → 404. Writes are atomic. Updating `app_tokens` takes effect immediately; updating `imei_map` is picked up by the listener within ~5 seconds | +| `GET /configs/{name}` | token | Read a config as JSON. `name`: `app_tokens` (secrets are not recoverable — app names with `null` placeholders) or `imei_map` (the full mapping) | +| `PUT/POST /configs/{name}` | token | Replace a config in the database. Body **must be a JSON object**; per-config schema is validated (imei_map keys must be 15-digit IMEIs; app_tokens must keep the `{"tokens": {name: secret}}` shape). Invalid payloads → 422, unknown names → 404. Writes are transactional (delete + insert). `app_tokens` changes take effect immediately; `imei_map` changes are picked up by the listener within ~5 seconds | | `GET /docs`, `GET /redoc` | none | OpenAPI docs (public) | +When the database is unreachable, authenticated endpoints return **503** (the API has no purpose without the db). + Example — everything device `Water tank sensor B` reported on August 31st, by device clock: ``` GET /messages?device=Water%20tank%20sensor%20B&payload_from=2026-08-31T00:00:00Z&payload_to=2026-08-31T23:59:59Z ``` -### Managing config files through the API +### Managing config through the API -The two JSON config files can be read and replaced over HTTP (authenticated like all other endpoints). This is how to add a device or rotate a token without touching the host filesystem: +Both configs live in the database (`app_tokens`, `imei_map` tables) and can be read and replaced over HTTP (authenticated like all other endpoints). This is how to add a device or rotate a token without direct database access: ```bash # read the current IMEI map -TOKEN='replace-me-with-a-long-random-secret-1' +TOKEN='your-bootstrap-or-app-token' curl -H "X-App-Token: $TOKEN" http://localhost:18080/configs/imei_map # replace it (JSON object only; keys must be 15-digit IMEIs) curl -X PUT -H "X-App-Token: $TOKEN" -H "Content-Type: application/json" \ - -d '{"350123456789012": "Greenhouse sensor A", "490154203237518": "Water tank sensor B"}' \ - http://localhost:18080/configs/imei_map + -d '{"350123456789012": "Greenhouse sensor A", "490154203237518": "Water tank sensor B"}' \ + http://localhost:18080/configs/imei_map ``` -- `app_tokens`: `{"tokens": {"app-name": "secret"}}` — after an update the API re-authenticates with the **new** tokens immediately. -- `imei_map`: `{"<15-digit-IMEI>": ""}` — the listener picks the change up within ~5 seconds (periodic reload); a temporarily invalid file on disk never breaks ingestion (the last good mapping stays in effect). +- `app_tokens`: `{"tokens": {"app-name": "secret"}}` — secrets are stored as SHA-256 digests (never plaintext) and cannot be read back; `GET /configs/app_tokens` shows only the app names. After an update the API re-authenticates with the **new** tokens immediately. +- `imei_map`: `{"<15-digit-IMEI>": ""}` — the listener polls the table every ~5 seconds; a temporarily unreachable database never breaks ingestion (the last good mapping stays in effect). + +#### Migrating an existing deployment from the JSON files + +For a database that already has data (init scripts only run on an empty volume), apply the schema and migrate the JSON configs: + +```bash +docker compose exec db psql -U nodeledge -d nodeledge -f - < init/01_schema.sql +python tools/migrate_configs_to_db.py --dsn postgresql://nodeledge:PASSWORD@localhost:5432/nodeledge \ + --tokens configs/api/app_tokens.json --imei-map configs/listener/imei_map.json +``` + +After migrating, the JSON files are no longer read by the stack and can be deleted. Example response: diff --git a/api/app.py b/api/app.py index 768e012..35368d9 100644 --- a/api/app.py +++ b/api/app.py @@ -1,20 +1,20 @@ """FastAPI application exposing read access to the mqtt_messages table and -management of the stack's JSON config files. +management of the config tables (app_tokens, imei_map) in the database. -Authentication: static app tokens from a JSON config file, passed either as -the `X-App-Token` header or as an `app_token` query parameter. Tokens are -compared in constant time; invalid tokens yield 401. +Authentication: app tokens stored as SHA-256 digests in the app_tokens +table, passed either as the `X-App-Token` header or as an `app_token` query +parameter. Digests are compared in constant time; invalid tokens yield 401. +When the database is unavailable, authenticated endpoints return 503. """ from __future__ import annotations +import hashlib import hmac -import json import logging import os -import tempfile +import threading from datetime import datetime, timezone -from pathlib import Path from typing import Any import psycopg @@ -26,39 +26,172 @@ from listener.process import validate_imei_map logger = logging.getLogger("api") -def _atomic_write(path: Path, content: str) -> None: - """Write content to path atomically: tempfile in the same directory, then - os.replace, so concurrent readers never observe a torn or empty file.""" - fd, tmp_name = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp") - try: - with os.fdopen(fd, "w", encoding="utf-8") as f: - f.write(content) - f.flush() - os.fsync(f.fileno()) - os.replace(tmp_name, path) - except Exception: - try: - os.unlink(tmp_name) - except OSError: - pass - raise +def hash_token(token: str) -> str: + """SHA-256 hex digest of an app token. Only digests are stored in the + database; plaintext tokens are never persisted.""" + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +def _db_unavailable(exc: Exception) -> HTTPException: + """503 for database outages: without the db the API has no purpose.""" + return HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=f"database unavailable: {exc}", + ) class AppConfig: """Runtime configuration loaded once at startup.""" - def __init__(self, dsn: str, tokens: dict[str, str], max_limit: int) -> None: + def __init__(self, dsn: str, max_limit: int) -> None: self.dsn = dsn - self.tokens = tokens self.max_limit = max_limit -def load_tokens(path: str) -> dict[str, str]: - """Load {"tokens": {"app-name": "secret"}} from a JSON file.""" - return _extract_tokens(json.loads(Path(path).read_text(encoding="utf-8")), path) +class TokenStore: + """API token store backed by the app_tokens table (SHA-256 digests). + + A bootstrap token (BOOTSTRAP_APP_TOKEN) is seeded lazily on the first + authentication attempt when the table is empty, so a fresh deployment + can authenticate without direct database access. Verification reads the + table on every request, so token updates take effect immediately. + """ + + def __init__( + self, + dsn: str, + bootstrap_token: str | None = None, + bootstrap_app: str = "bootstrap", + ) -> None: + self._dsn = dsn + self._bootstrap_token = bootstrap_token + self._bootstrap_app = bootstrap_app + self._bootstrap_lock = threading.Lock() + self._bootstrapped = False + + def _ensure_bootstrapped(self) -> None: + """Seed the bootstrap token once, if configured and the table is + empty. Runs lazily (on the first authenticate call) so a database + outage at startup does not prevent the API from starting; a failed + attempt is retried on the next request.""" + if self._bootstrapped: + return + with self._bootstrap_lock: + if self._bootstrapped: + return + conn = psycopg.connect(self._dsn, autocommit=True) + try: + with conn.cursor() as cur: + if self._bootstrap_token: + # Seed only when the table is empty; ON CONFLICT + # covers a race with another worker doing the same. + cur.execute( + """ + INSERT INTO app_tokens (app_name, token_hash) + SELECT %s, %s + WHERE NOT EXISTS (SELECT 1 FROM app_tokens) + ON CONFLICT DO NOTHING + """, + (self._bootstrap_app, hash_token(self._bootstrap_token)), + ) + if cur.rowcount: + logger.info( + "seeded bootstrap token %r from BOOTSTRAP_APP_TOKEN", + self._bootstrap_app, + ) + else: + cur.execute("SELECT count(*) FROM app_tokens") + if cur.fetchone()[0] == 0: + logger.warning( + "app_tokens is empty and BOOTSTRAP_APP_TOKEN is not " + "set; the API will reject all requests until a token " + "is added" + ) + finally: + conn.close() + self._bootstrapped = True + + def verify(self, token: str) -> bool: + """Constant-time check of a presented token against all digests.""" + self._ensure_bootstrapped() + digest = hash_token(token) + for stored in self._hashes().values(): + if hmac.compare_digest(stored, digest): + return True + return False + + def _hashes(self) -> dict[str, str]: + """All rows as {app_name: token_hash}.""" + conn = psycopg.connect(self._dsn) + try: + with conn.cursor() as cur: + cur.execute("SELECT app_name, token_hash FROM app_tokens") + return dict(cur.fetchall()) + finally: + conn.close() + + def names(self) -> list[str]: + """App names with a token (secrets are not recoverable).""" + conn = psycopg.connect(self._dsn) + try: + with conn.cursor() as cur: + cur.execute("SELECT app_name FROM app_tokens ORDER BY app_name") + return [row[0] for row in cur.fetchall()] + finally: + conn.close() + + def replace(self, tokens: dict[str, str]) -> None: + """Replace ALL tokens (delete + insert) in one transaction.""" + conn = psycopg.connect(self._dsn) + try: + with conn.transaction(): + with conn.cursor() as cur: + cur.execute("DELETE FROM app_tokens") + if tokens: + cur.executemany( + "INSERT INTO app_tokens (app_name, token_hash) " + "VALUES (%s, %s)", + [(name, hash_token(secret)) for name, secret in tokens.items()], + ) + finally: + conn.close() -def _extract_tokens(data: object, source: str) -> dict[str, str]: +class ImeiMapStore: + """IMEI -> device name mapping backed by the imei_map table.""" + + def __init__(self, dsn: str) -> None: + self._dsn = dsn + + def load(self) -> dict[str, str]: + conn = psycopg.connect(self._dsn) + try: + with conn.cursor() as cur: + cur.execute("SELECT imei, device FROM imei_map ORDER BY imei") + return {row[0]: row[1] for row in cur.fetchall()} + finally: + conn.close() + + def replace(self, mapping: dict[str, str]) -> None: + """Replace the whole mapping (delete + insert) in one transaction.""" + conn = psycopg.connect(self._dsn) + try: + with conn.transaction(): + with conn.cursor() as cur: + cur.execute("DELETE FROM imei_map") + if mapping: + cur.executemany( + "INSERT INTO imei_map (imei, device) VALUES (%s, %s)", + list(mapping.items()), + ) + finally: + conn.close() + + +def _validate_token_body(data: dict[str, Any], source: str) -> dict[str, str]: + """Validate a {"tokens": {app: secret}} request body and return the + inner tokens mapping. Raises ValueError with a descriptive message + otherwise.""" if not isinstance(data, dict) or not isinstance(data.get("tokens"), dict): raise ValueError(f"{source}: expected JSON on the form {{\"tokens\": {{name: secret}}}}") tokens = data["tokens"] @@ -66,7 +199,11 @@ def _extract_tokens(data: object, source: str) -> dict[str, str]: if not isinstance(secret, str) or not secret: raise ValueError(f"{source}: token {name!r} must be a non-empty string") if not tokens: - logger.warning("%s contains no tokens; the API will reject all requests", source) + logger.warning( + "%s replaces app_tokens with an empty set; the API will reject " + "all requests until a token is added again", + source, + ) return dict(tokens) @@ -84,30 +221,33 @@ def _as_utc(value: datetime) -> datetime: return value if value.tzinfo is not None else value.replace(tzinfo=timezone.utc) -def create_app() -> FastAPI: +def create_app( + token_store: TokenStore | None = None, + imei_store: ImeiMapStore | None = None, +) -> FastAPI: + """Build the API application. + + The stores are injectable so tests can run without a database; in + production the defaults are built from the environment (DB_DSN, + BOOTSTRAP_APP_TOKEN, BOOTSTRAP_APP_NAME). + """ app = FastAPI( title="nodeledge MQTT data API", version="1.0.0", description="Read access to MQTT-collected JSON payloads.", ) - tokens_path = _env("APP_TOKENS_FILE") dsn = _env("DB_DSN") max_limit = int(_env("API_MAX_LIMIT", "10000")) - config = AppConfig(dsn=dsn, tokens=load_tokens(tokens_path), max_limit=max_limit) - - # Whitelisted config files the /configs endpoints may read and write. Only - # these names resolve; anything else 404s so the API cannot touch files - # outside the mounted config directories. Validators enforce each file's - # schema on write (a JSON object plus per-config rules). - config_files: dict[str, Path] = { - "app_tokens": Path(tokens_path), - "imei_map": Path(_env("IMEI_MAP_FILE", "/app/config/imei_map.json")), - } - config_validators: dict[str, Any] = { - "app_tokens": lambda body: _extract_tokens(body, "request body"), - "imei_map": lambda body: validate_imei_map(body, "request body"), - } + config = AppConfig(dsn=dsn, max_limit=max_limit) + if token_store is None: + token_store = TokenStore( + dsn, + bootstrap_token=os.environ.get("BOOTSTRAP_APP_TOKEN") or None, + bootstrap_app=os.environ.get("BOOTSTRAP_APP_NAME") or "bootstrap", + ) + if imei_store is None: + imei_store = ImeiMapStore(dsn) def authenticate(request: Request) -> None: token = request.headers.get("X-App-Token") @@ -118,146 +258,83 @@ def create_app() -> FastAPI: status_code=status.HTTP_401_UNAUTHORIZED, detail="missing app token: pass the X-App-Token header", ) - for candidate in config.tokens.values(): - if hmac.compare_digest(candidate, token): - return - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="invalid app token", - ) + try: + if not token_store.verify(token): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="invalid app token", + ) + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc app.state.config = config app.state.authenticate = authenticate + # Whitelisted config names the /configs endpoints may touch; anything + # else 404s. Validators enforce each table's schema on write. + config_names = ("app_tokens", "imei_map") + + def _unknown_config(name: str) -> HTTPException: + return HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"unknown config {name!r}; available: {', '.join(config_names)}", + ) + @app.get("/configs/{name}", tags=["config"], dependencies=[Depends(authenticate)]) def get_config(name: str) -> dict[str, Any]: - path = config_files.get(name) - if path is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"unknown config {name!r}; available: {', '.join(sorted(config_files))}", - ) try: - content = path.read_text(encoding="utf-8") - except FileNotFoundError: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"config file {name!r} not found on disk", - ) - try: - return json.loads(content) - except json.JSONDecodeError as exc: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"config file {name!r} contains invalid JSON: {exc}", - ) - - def _set_config(name: str, body: dict[str, Any]) -> dict[str, Any]: - path = config_files.get(name) - if path is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"unknown config {name!r}; available: {', '.join(sorted(config_files))}", - ) - validator = config_validators.get(name) - if validator is not None: - try: - validator(body) - except ValueError as exc: - raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, - detail=str(exc), - ) - try: - _atomic_write(path, json.dumps(body, indent=2, ensure_ascii=False) + "\n") - except OSError as exc: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"failed to write config {name!r}: {exc}", - ) - # app_tokens backs this API's own authentication: reload immediately - # so callers are not stuck with the old tokens after an update. - if name == "app_tokens": - config.tokens = load_tokens(str(path)) - logger.info("app tokens reloaded after config update") - return {"status": "updated", "config": name} + if name == "app_tokens": + # Secrets are stored as digests and not recoverable; expose + # only the app names so clients can see what exists. + return {"tokens": {app: None for app in token_store.names()}} + if name == "imei_map": + return imei_store.load() + raise _unknown_config(name) + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc @app.put("/configs/{name}", tags=["config"], dependencies=[Depends(authenticate)]) @app.post("/configs/{name}", tags=["config"], dependencies=[Depends(authenticate)]) def set_config(name: str, body: dict[str, Any]) -> dict[str, Any]: - return _set_config(name, body) - - @app.get("/configs/{name}", tags=["config"], dependencies=[Depends(authenticate)]) - def get_config(name: str) -> dict[str, Any]: - path = config_files.get(name) - if path is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"unknown config {name!r}; available: {', '.join(sorted(config_files))}", - ) try: - content = path.read_text(encoding="utf-8") - except FileNotFoundError: + if name == "app_tokens": + tokens = _validate_token_body(body, "request body") + token_store.replace(tokens) + # The store reads the table on every request, so the new + # tokens take effect immediately - no reload needed. + logger.info("app_tokens replaced: %d token(s)", len(tokens)) + elif name == "imei_map": + mapping = validate_imei_map(body, "request body") + imei_store.replace(mapping) + # The listener polls the imei_map table every 5 seconds. + logger.info("imei_map replaced: %d mapping(s)", len(mapping)) + else: + raise _unknown_config(name) + except ValueError as exc: raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"config file {name!r} not found on disk", - ) - try: - return json.loads(content) - except json.JSONDecodeError as exc: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"config file {name!r} contains invalid JSON: {exc}", - ) - - def _set_config(name: str, body: dict[str, Any]) -> dict[str, Any]: - path = config_files.get(name) - if path is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"unknown config {name!r}; available: {', '.join(sorted(config_files))}", - ) - validator = config_validators.get(name) - if validator is not None: - try: - validator(body) - except ValueError as exc: - raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, - detail=str(exc), - ) - try: - _atomic_write(path, json.dumps(body, indent=2, ensure_ascii=False) + "\n") - except OSError as exc: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"failed to write config {name!r}: {exc}", - ) - # app_tokens backs this API's own authentication: reload immediately - # so callers are not stuck with the old tokens after an update. - if name == "app_tokens": - config.tokens = load_tokens(str(path)) - logger.info("app tokens reloaded after config update") + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=str(exc), + ) from exc + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc return {"status": "updated", "config": name} - @app.put("/configs/{name}", tags=["config"], dependencies=[Depends(authenticate)]) - @app.post("/configs/{name}", tags=["config"], dependencies=[Depends(authenticate)]) - def set_config(name: str, body: dict[str, Any]) -> dict[str, Any]: - return _set_config(name, body) - @app.get("/health", tags=["meta"]) def health() -> dict[str, Any]: return {"status": "ok"} @app.get("/messages/count", tags=["meta"], dependencies=[Depends(authenticate)]) def message_count() -> dict[str, int]: - conn = psycopg.connect(config.dsn) try: - with conn.cursor() as cur: - cur.execute("SELECT count(*) FROM mqtt_messages") - row = cur.fetchone() - finally: - conn.close() + conn = psycopg.connect(config.dsn) + try: + with conn.cursor() as cur: + cur.execute("SELECT count(*) FROM mqtt_messages") + row = cur.fetchone() + finally: + conn.close() + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc count = row[0] if row is not None else 0 return {"count": count} @@ -353,13 +430,16 @@ def create_app() -> FastAPI: "SELECT id, received_at, topic, imei, device, payload, payload_time FROM mqtt_messages" ) + where + order_by - conn = psycopg.connect(config.dsn) try: - with conn.cursor() as cur: - cur.execute(query, (*params, limit, offset)) - rows = cur.fetchall() - finally: - conn.close() + conn = psycopg.connect(config.dsn) + try: + with conn.cursor() as cur: + cur.execute(query, (*params, limit, offset)) + rows = cur.fetchall() + finally: + conn.close() + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc return { "limit": limit, diff --git a/docker-compose.yml b/docker-compose.yml index efa4e7f..85e4aa2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,9 +1,10 @@ # Nodeledge MQTT listener stack (Portainer / docker compose compatible). # # Services: -# db PostgreSQL holding the mqtt_messages table +# db PostgreSQL holding the mqtt_messages table + the config +# tables (app_tokens, imei_map) # mqtt-listener subscribes to MQTT, injects IMEI + device into JSON payloads, writes rows -# api FastAPI read API over the database, protected by static app tokens +# api FastAPI read API over the database, protected by app tokens # broker OPTIONAL test broker (compose profile "test") for trying the stack out # # Portainer: Stacks -> Add stack -> paste this file, then set the environment @@ -49,16 +50,12 @@ services: MQTT_TOPIC_FILTER: ${MQTT_TOPIC_FILTER:-/+/+} MQTT_QOS: ${MQTT_QOS:-1} DB_DSN: postgresql://${POSTGRES_USER:-nodeledge}:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in the stack environment}@db:5432/${POSTGRES_DB:-nodeledge} - IMEI_MAP_FILE: /app/config/imei_map.json INJECT_IMEI_KEY: ${INJECT_IMEI_KEY:-imei} INJECT_DEVICE_KEY: ${INJECT_DEVICE_KEY:-device} # Comma-separated payload keys searched for a device timestamp. PAYLOAD_TIME_KEYS: ${PAYLOAD_TIME_KEYS:-timestamp,time,ts} LOG_LEVEL: ${LOG_LEVEL:-INFO} TZ: ${TZ:-Europe/Stockholm} - volumes: - # IMEI -> device name mapping, see configs/listener/imei_map.json - - ./configs/listener:/app/config:ro healthcheck: test: ["CMD", "python", "/app/listener/healthcheck.py"] interval: 30s @@ -77,18 +74,15 @@ services: condition: service_healthy environment: DB_DSN: postgresql://${POSTGRES_USER:-nodeledge}:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in the stack environment}@db:5432/${POSTGRES_DB:-nodeledge} - APP_TOKENS_FILE: /app/config/app_tokens.json - # Read/write location for the listener's IMEI map, shared via the - # ./configs/listener mount (the listener reads the same file) - IMEI_MAP_FILE: /app/config-listener/imei_map.json + # Seed token for the FIRST deployment: on the first authenticated + # request the api inserts its digest into the (then empty) app_tokens + # table. Rotate to real tokens via PUT /configs/app_tokens, then unset + # this variable. Ignored once app_tokens has rows. + BOOTSTRAP_APP_TOKEN: ${BOOTSTRAP_APP_TOKEN:-} + BOOTSTRAP_APP_NAME: ${BOOTSTRAP_APP_NAME:-bootstrap} API_MAX_LIMIT: ${API_MAX_LIMIT:-10000} LOG_LEVEL: ${LOG_LEVEL:-INFO} TZ: ${TZ:-Europe/Stockholm} - volumes: - # app tokens for API auth + the listener's IMEI map; mounted rw so the - # /configs endpoints can update them (writes are atomic) - - ./configs/api:/app/config:rw - - ./configs/listener:/app/config-listener:rw ports: - "${API_HOST_PORT:-18080}:8000" healthcheck: diff --git a/init/01_schema.sql b/init/01_schema.sql index 7c6f0b7..24cac38 100644 --- a/init/01_schema.sql +++ b/init/01_schema.sql @@ -1,6 +1,9 @@ -- Schema for the nodeledge MQTT data pipeline. -- Executed automatically by the postgres image on the FIRST database start -- (files in /docker-entrypoint-initdb.d are only run against an empty data dir). +-- For an ALREADY RUNNING database apply the changes manually, e.g. +-- docker compose exec db psql -U nodeledge -d nodeledge -f - < init/01_schema.sql +-- (CREATE TABLE IF NOT EXISTS makes the whole script idempotent). CREATE TABLE IF NOT EXISTS mqtt_messages ( id BIGSERIAL PRIMARY KEY, @@ -32,4 +35,25 @@ COMMENT ON COLUMN mqtt_messages.payload_time IS 'Device timestamp extracted from the JSON payload at ingest time; NULL when absent or unparseable. received_at is the server receipt time.'; COMMENT ON TABLE mqtt_messages IS - 'JSON payloads received over MQTT; payload contains the original JSON with imei and device injected by the listener.'; \ No newline at end of file + 'JSON payloads received over MQTT; payload contains the original JSON with imei and device injected by the listener.'; + +-- API authentication tokens, stored as hex SHA-256 digests (never plaintext). +-- The api seeds one row from BOOTSTRAP_APP_TOKEN when the table is empty. +CREATE TABLE IF NOT EXISTS app_tokens ( + app_name TEXT PRIMARY KEY, + token_hash TEXT NOT NULL CHECK (token_hash ~ '^[0-9a-f]{64}$'), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- IMEI -> device name mapping used by the listener at ingest time. +CREATE TABLE IF NOT EXISTS imei_map ( + imei TEXT PRIMARY KEY CHECK (imei ~ '^[0-9]{15}$'), + device TEXT NOT NULL CHECK (device <> ''), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +COMMENT ON TABLE app_tokens IS + 'API app tokens as SHA-256 hex digests; seeded from BOOTSTRAP_APP_TOKEN when empty.'; +COMMENT ON TABLE imei_map IS + 'IMEI -> device name mapping, polled by the listener every 5 seconds.'; \ No newline at end of file diff --git a/listener/db.py b/listener/db.py index 33f1171..bcb8d8f 100644 --- a/listener/db.py +++ b/listener/db.py @@ -2,6 +2,9 @@ The network thread writes rows; writes happen only on that thread so a plain (no pool) blocking connection is used, with reconnection on failure. + +The IMEI -> device mapping lives in the imei_map table and is polled by a +separate daemon thread (ImeiMapReader) with its own connection. """ from __future__ import annotations @@ -78,6 +81,87 @@ class Inserter: self._conn.close() except psycopg.Error: pass + + +class ImeiMapReader: + """Thread that polls the imei_map table for the IMEI -> device mapping. + + Runs its OWN persistent autocommit connection (never shared with + Inserter, whose connection is funneled through the MQTT network thread). + Reconnects on failure; a failed poll keeps the previously read mapping, + mirroring the old broken-file-keeps-last-good-map semantics. + """ + + POLL_INTERVAL_SECONDS = 5.0 + + def __init__(self, dsn: str) -> None: + self._dsn = dsn + self._lock = threading.Lock() + self._conn: Connection | None = None + self._imap: dict[str, str] = {} + self._stop = threading.Event() + self._thread = threading.Thread( + target=self._run, name="imei-map-poll", daemon=True + ) + + def _connection(self) -> Connection: + if self._conn is None or self._conn.closed: + logger.info("connecting to database (imei map poll)") + self._conn = psycopg.connect(self._dsn, autocommit=True) + return self._conn + + def read(self) -> dict[str, str]: + """Read the mapping from the database (single SELECT). + + Raises psycopg.Error on connection/query failure; the caller keeps + the last good mapping.""" + conn = self._connection() + with conn.cursor() as cur: + cur.execute("SELECT imei, device FROM imei_map") + return dict(cur.fetchall()) + + def _run(self) -> None: + """Poll loop; runs in its own daemon thread.""" + while not self._stop.wait(timeout=self.POLL_INTERVAL_SECONDS): + try: + fresh = self.read() + except psycopg.Error: + logger.exception("IMEI map poll failed; keeping previous mapping") + self._close() + continue + with self._lock: + if fresh != self._imap: + self._imap = fresh + logger.info("IMEI map updated: %d mappings", len(fresh)) + + def start(self) -> None: + self._thread.start() + + def prime(self, mapping: dict[str, str]) -> None: + """Set the initial mapping (call before start()).""" + with self._lock: + self._imap = dict(mapping) + + def snapshot(self) -> dict[str, str]: + """Thread-safe copy of the current mapping for the ingest path.""" + with self._lock: + return dict(self._imap) + + def _close(self) -> None: + if self._conn is not None: + try: + self._conn.close() + except psycopg.Error: + pass + self._conn = None + + def close(self) -> None: + """Stop the poll thread and close the connection.""" + self._stop.set() + if self._thread.is_alive(): + self._thread.join(timeout=self.POLL_INTERVAL_SECONDS) + with self._lock: + self._close() self._conn = None def close(self) -> None: diff --git a/listener/main.py b/listener/main.py index 27a9938..6740a63 100644 --- a/listener/main.py +++ b/listener/main.py @@ -9,7 +9,6 @@ may be retried or dropped depending on QoS. from __future__ import annotations -import json import logging import os import signal @@ -17,10 +16,11 @@ import threading from pathlib import Path import paho.mqtt.client as mqtt +import psycopg from paho.mqtt.enums import CallbackAPIVersion -from listener.db import Inserter -from listener.process import DEFAULT_TIME_KEYS, normalize_imei, process_message, validate_imei_map +from listener.db import ImeiMapReader, Inserter +from listener.process import DEFAULT_TIME_KEYS, process_message logger = logging.getLogger("listener") @@ -28,15 +28,6 @@ _stop_event = threading.Event() _shutdown_complete = threading.Event() -def load_imei_map(path: str, log: bool = True) -> dict[str, str]: - """Load and validate the IMEI -> device name map from a JSON file.""" - data = json.loads(Path(path).read_text(encoding="utf-8")) - result = validate_imei_map(data, path) - if log: - logger.info("loaded %d IMEI mappings from %s", len(result), path) - return result - - def _env(name: str, default: str | None = None) -> str: value = os.environ.get(name) if not value: @@ -61,9 +52,17 @@ def main() -> None: topic_filter = _env("MQTT_TOPIC_FILTER", "/+/+") qos = int(_env("MQTT_QOS", "1")) - map_file = _env("IMEI_MAP_FILE") - imei_map = load_imei_map(map_file) - imei_lock = threading.Lock() # guards imei_map against the reload thread + map_reader = ImeiMapReader(_env("DB_DSN")) + # Fail fast when the db (or the imei_map table) is unreachable: without a + # mapping the listener would silently store every device name as its IMEI. + try: + initial_map = map_reader.read() + except psycopg.Error: + logger.exception("initial IMEI map read failed") + raise + map_reader.prime(initial_map) + map_reader.start() + logger.info("loaded %d IMEI mappings from the database", len(initial_map)) imei_key = _env("INJECT_IMEI_KEY", "imei") device_key = _env("INJECT_DEVICE_KEY", "device") time_keys = tuple( @@ -78,25 +77,6 @@ def main() -> None: Path("/tmp/listener.pid").write_text(str(os.getpid()), encoding="utf-8") logger.info("listener starting, pid %d", os.getpid()) - def reload_imei_map() -> None: - """Periodically reload the IMEI map so edits via the API take effect - without a container restart. A broken file is logged and skipped; the - previously loaded mapping stays in effect.""" - while not _stop_event.wait(timeout=5.0): - try: - fresh = load_imei_map(map_file, log=False) - except (OSError, ValueError, json.JSONDecodeError): - logger.exception("IMEI map reload failed; keeping previous mapping") - continue - nonlocal imei_map - with imei_lock: - if fresh != imei_map: - imei_map = fresh - logger.info("IMEI map reloaded: %d mappings", len(fresh)) - - reload_thread = threading.Thread(target=reload_imei_map, name="imei-map-reload", daemon=True) - reload_thread.start() - def on_connect(client, userdata, flags, reason_code, properties): # noqa: ANN001 if reason_code.is_failure: logger.error("broker refused connection: %s", reason_code) @@ -107,12 +87,10 @@ def main() -> None: def on_message(client, userdata, message): # noqa: ANN001 try: topic = message.topic - with imei_lock: - current_map = imei_map results = process_message( topic, bytes(message.payload), - current_map, + map_reader.snapshot(), imei_key=imei_key, device_key=device_key, time_keys=time_keys, @@ -157,6 +135,7 @@ def main() -> None: client.loop_stop() client.disconnect() inserter.close() + map_reader.close() Path("/tmp/listener.pid").unlink(missing_ok=True) logger.info("shutdown complete") diff --git a/tools/migrate_configs_to_db.py b/tools/migrate_configs_to_db.py new file mode 100644 index 0000000..0c65cea --- /dev/null +++ b/tools/migrate_configs_to_db.py @@ -0,0 +1,115 @@ +"""One-time migration: app_tokens.json + imei_map.json -> database tables. + +Reads the legacy JSON config files, inserts app tokens as SHA-256 digests +into app_tokens and IMEI -> device mappings into imei_map, then optionally +verifies the migration with a login round-trip. Run it once against an +existing deployment after applying init/01_schema.sql (the script is +idempotent: it truncates both tables first). + +Usage: + python tools/migrate_configs_to_db.py --dsn postgresql://user:pass@host:5432/db \\ + --tokens configs/api/app_tokens.json --imei-map configs/listener/imei_map.json \\ + [--api-url http://localhost:18080] [--api-token ] + +Requires psycopg (see tools/requirements.txt). Set --api-url/--api-token to +verify that the running API authenticates against the migrated tokens. +""" + +from __future__ import annotations + +import argparse +import json +import sys +import urllib.error +import urllib.request +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import psycopg # noqa: E402 + +from api.app import hash_token # noqa: E402 +from listener.process import validate_imei_map # noqa: E402 + + +def load_tokens(path: str) -> dict[str, str]: + data = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(data, dict) or not isinstance(data.get("tokens"), dict): + raise SystemExit(f"{path}: expected JSON on the form {{\"tokens\": {{name: secret}}}}") + tokens = data["tokens"] + for name, secret in tokens.items(): + if not isinstance(secret, str) or not secret: + raise SystemExit(f"{path}: token {name!r} must be a non-empty string") + return dict(tokens) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--dsn", required=True, help="PostgreSQL DSN of the stack's db") + parser.add_argument("--tokens", required=True, help="legacy app_tokens.json path") + parser.add_argument("--imei-map", required=True, help="legacy imei_map.json path") + parser.add_argument("--api-url", default=None, help="base URL of the running API for verification") + parser.add_argument( + "--api-token", + default=None, + help="one plaintext token from the migrated file, used for verification", + ) + args = parser.parse_args() + + tokens = load_tokens(args.tokens) + mapping = validate_imei_map( + json.loads(Path(args.imei_map).read_text(encoding="utf-8")), args.imei_map + ) + + print(f"migrating {len(tokens)} token(s) and {len(mapping)} IMEI mapping(s)") + with psycopg.connect(args.dsn) as conn: + with conn.transaction(): + with conn.cursor() as cur: + cur.execute("DELETE FROM app_tokens") + cur.execute("DELETE FROM imei_map") + cur.executemany( + "INSERT INTO app_tokens (app_name, token_hash) VALUES (%s, %s)", + [(name, hash_token(secret)) for name, secret in tokens.items()], + ) + cur.executemany( + "INSERT INTO imei_map (imei, device) VALUES (%s, %s)", + list(mapping.items()), + ) + + print("migrated. verifying read-back...") + with psycopg.connect(args.dsn) as conn: + with conn.cursor() as cur: + cur.execute("SELECT count(*) FROM app_tokens") + row = cur.fetchone() + token_count = row[0] if row is not None else 0 + cur.execute("SELECT count(*) FROM imei_map") + row = cur.fetchone() + map_count = row[0] if row is not None else 0 + if token_count != len(tokens) or map_count != len(mapping): + print(f"FAIL: expected {len(tokens)}/{len(mapping)}, found {token_count}/{map_count}") + return 1 + print(f"PASS: app_tokens has {token_count} row(s), imei_map has {map_count} row(s)") + + if args.api_url: + if not args.api_token: + print("--api-token is required with --api-url") + return 1 + req = urllib.request.Request(f"{args.api_url}/messages?limit=1") + req.add_header("X-App-Token", args.api_token) + try: + with urllib.request.urlopen(req, timeout=10) as response: + status = response.status + except urllib.error.HTTPError as exc: + status = exc.code + if status == 200: + print("PASS: the running API accepts the migrated token") + else: + print(f"FAIL: API returned {status} for the migrated token (restart the api?)") + return 1 + + print("done. the JSON files are no longer read by the stack; delete them at will.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/tools/test_listener.py b/tools/test_listener.py index 76463f9..f9fc748 100644 --- a/tools/test_listener.py +++ b/tools/test_listener.py @@ -1,8 +1,9 @@ -"""Unit tests for listener.process and listener.main config loading.""" +"""Unit tests for listener.process and the API's config/token stores.""" from __future__ import annotations import json +import os import sys from pathlib import Path @@ -10,11 +11,19 @@ import pytest sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -from listener.main import load_imei_map # noqa: E402 +# api.app runs create_app() at import time, which requires DB_DSN. It never +# connects at import (connections are lazy), so a fake DSN is enough; the +# app_module fixture re-creates the app with injectable in-memory stores. +os.environ.setdefault( + "DB_DSN", "postgresql://user:pass@localhost:5432/db?connect_timeout=2" +) + +from api.app import ImeiMapStore, TokenStore, _validate_token_body, hash_token # noqa: E402 from listener.process import ( # noqa: E402 InjectionResult, parse_payload_time, process_message, + validate_imei_map, ) from datetime import datetime, timezone # noqa: E402 @@ -256,112 +265,174 @@ class TestPayloadTimeExtraction: assert result[0].payload_time is None -class TestLoadImeiMap: - def test_loads_valid_map(self, tmp_path: Path) -> None: - path = tmp_path / "map.json" - path.write_text( - json.dumps({"350123456789012": "Device X"}), encoding="utf-8" - ) - loaded = load_imei_map(str(path)) - assert loaded == {"350123456789012": "Device X"} +class TestValidateImeiMap: + """validate_imei_map is the shared schema check for the imei_map table + writes (API request bodies and the migration tool).""" - def test_rejects_non_object(self, tmp_path: Path) -> None: - path = tmp_path / "map.json" - path.write_text("[]", encoding="utf-8") + def test_accepts_valid_map(self) -> None: + assert validate_imei_map( + {"350123456789012": "Device X"}, "test" + ) == {"350123456789012": "Device X"} + + def test_rejects_non_object(self) -> None: with pytest.raises(ValueError, match="expected a JSON object"): - load_imei_map(str(path)) + validate_imei_map([], "test") - def test_rejects_non_string_values(self, tmp_path: Path) -> None: - path = tmp_path / "map.json" - path.write_text(json.dumps({"350123456789012": 42}), encoding="utf-8") + def test_rejects_non_string_values(self) -> None: with pytest.raises(ValueError, match="non-empty strings"): - load_imei_map(str(path)) + validate_imei_map({"350123456789012": 42}, "test") - def test_rejects_non_imei_key(self, tmp_path: Path) -> None: - path = tmp_path / "map.json" + def test_rejects_non_imei_key(self) -> None: # Old dev EUI-style key: no longer accepted - path.write_text(json.dumps({"0011223344556677": "Device X"}), encoding="utf-8") with pytest.raises(ValueError, match="15-digit IMEI"): - load_imei_map(str(path)) + validate_imei_map({"0011223344556677": "Device X"}, "test") + + +class TestTokenHelpers: + def test_hash_token_is_sha256_hex(self) -> None: + digest = hash_token("secret-abc") + assert len(digest) == 64 + int(digest, 16) # hex digest + + def test_hash_token_is_deterministic(self) -> None: + assert hash_token("secret-abc") == hash_token("secret-abc") + assert hash_token("secret-abc") != hash_token("secret-abd") + + def test_validate_token_body_extracts_tokens(self) -> None: + tokens = _validate_token_body({"tokens": {"app": "secret"}}, "test") + assert tokens == {"app": "secret"} + + def test_validate_token_body_rejects_wrong_shape(self) -> None: + with pytest.raises(ValueError, match="expected JSON on the form"): + _validate_token_body({"foo": "bar"}, "test") + + def test_validate_token_body_rejects_empty_secret(self) -> None: + with pytest.raises(ValueError, match="non-empty string"): + _validate_token_body({"tokens": {"app": ""}}, "test") + + +class FakeTokenStore(TokenStore): + """In-memory token store: no database, immediate verification.""" + + def __init__(self, tokens: dict[str, str]) -> None: + super().__init__("postgresql://unused") + self._tokens = dict(tokens) + self.replaced_with: dict[str, str] | None = None + + def verify(self, token: str) -> bool: + return token in self._tokens.values() + + def names(self) -> list[str]: + return sorted(self._tokens) + + def replace(self, tokens: dict[str, str]) -> None: + self.replaced_with = dict(tokens) + self._tokens = dict(tokens) + + +class FakeImeiMapStore(ImeiMapStore): + """In-memory imei map store: no database.""" + + def __init__(self, mapping: dict[str, str]) -> None: + super().__init__("postgresql://unused") + self._mapping = dict(mapping) + self.replaced_with: dict[str, str] | None = None + + def load(self) -> dict[str, str]: + return dict(self._mapping) + + def replace(self, mapping: dict[str, str]) -> None: + self.replaced_with = dict(mapping) + self._mapping = dict(mapping) @pytest.fixture() -def app_module(tmp_path): - """Import api.app with a prepared environment (create_app runs at import - time). Yields the reloaded module and restores the environment after.""" +def app_module(monkeypatch): + """Import api.app with a fake environment (create_app runs at import + time; DB connections are lazy so nothing connects yet). monkeypatch + restores the environment automatically.""" import importlib - import os - # connect_timeout keeps the DB-connecting tests fast even when + # Legacy file-based env vars must NOT be set for the reloaded module. + for legacy in ("APP_TOKENS_FILE", "IMEI_MAP_FILE"): + monkeypatch.delenv(legacy, raising=False) + # connect_timeout keeps DB-connecting tests fast even when # localhost:5432 is silently dropped by a firewall (psycopg otherwise # waits out the OS-level TCP timeout). - tokens = tmp_path / "app_tokens.json" - tokens.write_text( - json.dumps({"tokens": {"test-app": "secret-abc"}}), encoding="utf-8" + monkeypatch.setenv( + "DB_DSN", "postgresql://user:pass@localhost:5432/db?connect_timeout=2" ) - imei_map = tmp_path / "imei_map.json" - imei_map.write_text( - json.dumps({"350123456789012": "Greenhouse sensor A"}), encoding="utf-8" - ) - env = { - "APP_TOKENS_FILE": str(tokens), - "IMEI_MAP_FILE": str(imei_map), - "DB_DSN": "postgresql://user:pass@localhost:5432/db?connect_timeout=2", - } - old = {key: os.environ.get(key) for key in env} - os.environ.update(env) - try: - import api.app as app_module + import api.app as app_module - importlib.reload(app_module) - yield app_module - finally: - for key, value in old.items(): - if value is None: - os.environ.pop(key, None) - else: - os.environ[key] = value + importlib.reload(app_module) + yield app_module + + +@pytest.fixture() +def api(app_module): + """ASGI app wired to in-memory fakes; yields (app, token_store, imei_store).""" + tokens = FakeTokenStore({"test-app": "secret-abc"}) + imeis = FakeImeiMapStore({"350123456789012": "Greenhouse sensor A"}) + app = app_module.create_app(token_store=tokens, imei_store=imeis) + return app, tokens, imeis class TestApiAuth: """Light tests of the API auth dependency using the ASGI app directly.""" - def test_health_is_public(self, app_module) -> None: + def test_health_is_public(self, api) -> None: from fastapi.testclient import TestClient - client = TestClient(app_module.app) - assert client.get("/health").status_code == 405 or client.get("/health").json() == {"status": "ok"} + client = TestClient(api[0]) + assert client.get("/health").json() == {"status": "ok"} - def test_missing_token_rejected(self, app_module) -> None: + def test_missing_token_rejected(self, api) -> None: from fastapi.testclient import TestClient - client = TestClient(app_module.app) + client = TestClient(api[0]) response = client.get("/messages?limit=1") assert response.status_code == 401 - def test_wrong_token_rejected(self, app_module) -> None: + def test_wrong_token_rejected(self, api) -> None: from fastapi.testclient import TestClient - client = TestClient(app_module.app) + client = TestClient(api[0]) response = client.get("/messages?limit=1", headers={"X-App-Token": "nope"}) assert response.status_code == 401 - def test_valid_token_accepted(self, app_module) -> None: + def test_valid_token_accepted(self, api) -> None: from fastapi.testclient import TestClient # The DB is not reachable in unit tests, so a valid token proceeds past # auth and then fails on the database connection with a 500. That # proves authentication passed (401 would mean it did not). - client = TestClient(app_module.app, raise_server_exceptions=False) + client = TestClient(api[0], raise_server_exceptions=False) response = client.get("/messages?limit=1", headers={"X-App-Token": "secret-abc"}) assert response.status_code != 401 - def test_reversed_time_frame_rejected(self, app_module) -> None: + def test_db_outage_yields_503_not_500(self, app_module) -> None: + from fastapi.testclient import TestClient + + import psycopg + + class BrokenStore(FakeTokenStore): + def verify(self, token: str) -> bool: + raise psycopg.OperationalError("db is down") + + app = app_module.create_app( + token_store=BrokenStore({"test-app": "secret-abc"}), + imei_store=FakeImeiMapStore({}), + ) + client = TestClient(app, raise_server_exceptions=False) + response = client.get("/messages?limit=1", headers={"X-App-Token": "secret-abc"}) + assert response.status_code == 503 + + def test_reversed_time_frame_rejected(self, api) -> None: from fastapi.testclient import TestClient # Range validation runs before any database access, so this works # without a reachable DB. - client = TestClient(app_module.app) + client = TestClient(api[0]) response = client.get( "/messages", params={"from": "2026-08-31T12:00:00Z", "to": "2026-08-31T11:00:00Z"}, @@ -369,12 +440,12 @@ class TestApiAuth: ) assert response.status_code == 422 - def test_naive_and_offset_timestamps_accepted(self, app_module) -> None: + def test_naive_and_offset_timestamps_accepted(self, api) -> None: from fastapi.testclient import TestClient # Both forms must parse; the request fails later on DB access (500), # never with a 422 validation error. - client = TestClient(app_module.app, raise_server_exceptions=False) + client = TestClient(api[0], raise_server_exceptions=False) for value in ("2026-08-31T12:00:00", "2026-08-31T12:00:00+02:00"): response = client.get( "/messages", @@ -385,85 +456,96 @@ class TestApiAuth: class TestConfigEndpoints: - """GET/PUT/POST /configs/{name} — reads, validated JSON writes.""" - - @pytest.fixture() - def client(self, app_module): - from fastapi.testclient import TestClient - - return TestClient(app_module.app) + """GET/PUT/POST /configs/{name} — reads, validated writes into stores.""" HDRS = {"X-App-Token": "secret-abc"} - def test_get_imei_map(self, client, app_module) -> None: + @pytest.fixture() + def client_and_stores(self, api): + from fastapi.testclient import TestClient + + app, tokens, imeis = api + return TestClient(app), tokens, imeis + + def test_get_imei_map(self, client_and_stores) -> None: + client, _, imeis = client_and_stores r = client.get("/configs/imei_map", headers=self.HDRS) assert r.status_code == 200 assert r.json() == {"350123456789012": "Greenhouse sensor A"} + assert imeis.load() == {"350123456789012": "Greenhouse sensor A"} - def test_get_app_tokens(self, client) -> None: + def test_get_app_tokens_masks_secrets(self, client_and_stores) -> None: + # secrets are stored as digests and not recoverable; only names show + client, _, _ = client_and_stores r = client.get("/configs/app_tokens", headers=self.HDRS) assert r.status_code == 200 - assert r.json() == {"tokens": {"test-app": "secret-abc"}} + assert r.json() == {"tokens": {"test-app": None}} - def test_get_unknown_config_404(self, client) -> None: + def test_get_unknown_config_404(self, client_and_stores) -> None: + client, _, _ = client_and_stores r = client.get("/configs/does-not-exist", headers=self.HDRS) assert r.status_code == 404 - def test_config_endpoints_require_token(self, client) -> None: + def test_config_endpoints_require_token(self, client_and_stores) -> None: + client, _, _ = client_and_stores assert client.get("/configs/imei_map").status_code == 401 assert client.put("/configs/imei_map", json={}).status_code == 401 - def test_put_valid_imei_map(self, client, app_module) -> None: - import os - + def test_put_valid_imei_map(self, client_and_stores) -> None: + client, _, imeis = client_and_stores body = {"350123456789012": "Greenhouse sensor A", "490154203237518": "Water tank sensor B"} r = client.put("/configs/imei_map", json=body, headers=self.HDRS) assert r.status_code == 200 assert r.json() == {"status": "updated", "config": "imei_map"} - # file on disk holds the new content - with open(os.environ["IMEI_MAP_FILE"], encoding="utf-8") as f: - assert json.load(f) == body + # the store (and thereby the table) holds the new content + assert imeis.replaced_with == body - def test_post_is_alias_for_put(self, client) -> None: + def test_post_is_alias_for_put(self, client_and_stores) -> None: + client, _, _ = client_and_stores body = {"350123456789012": "Greenhouse sensor A"} r = client.post("/configs/imei_map", json=body, headers=self.HDRS) assert r.status_code == 200 - def test_put_rejects_non_imei_keys(self, client) -> None: + def test_put_rejects_non_imei_keys(self, client_and_stores) -> None: + client, _, _ = client_and_stores r = client.put("/configs/imei_map", json={"0011223344556677": "X"}, headers=self.HDRS) assert r.status_code == 422 assert "15-digit IMEI" in r.json()["detail"] - def test_put_rejects_non_string_values(self, client) -> None: + def test_put_rejects_non_string_values(self, client_and_stores) -> None: + client, _, _ = client_and_stores r = client.put("/configs/imei_map", json={"350123456789012": 42}, headers=self.HDRS) assert r.status_code == 422 - def test_put_rejects_bad_token_file_shape(self, client) -> None: + def test_put_rejects_bad_token_body_shape(self, client_and_stores) -> None: # JSON object, but not the {"tokens": {...}} shape + client, _, _ = client_and_stores r = client.put("/configs/app_tokens", json={"foo": "bar"}, headers=self.HDRS) assert r.status_code == 422 - def test_put_rejects_empty_token_secret(self, client) -> None: + def test_put_rejects_empty_token_secret(self, client_and_stores) -> None: + client, _, _ = client_and_stores r = client.put( "/configs/app_tokens", json={"tokens": {"app": ""}}, headers=self.HDRS ) assert r.status_code == 422 - def test_non_object_body_rejected_by_fastapi(self, client) -> None: + def test_non_object_body_rejected_by_fastapi(self, client_and_stores) -> None: # JSON but not an object: FastAPI's dict[str, Any] annotation rejects # arrays and scalars before the handler runs + client, _, _ = client_and_stores r = client.put("/configs/imei_map", json=[1, 2, 3], headers=self.HDRS) assert r.status_code == 422 r = client.put("/configs/imei_map", json="just a string", headers=self.HDRS) assert r.status_code == 422 - def test_app_tokens_update_reloads_auth(self, client, app_module) -> None: + def test_app_tokens_update_reloads_auth(self, client_and_stores) -> None: # rotate the token, then authenticate with the NEW one + client, tokens, _ = client_and_stores new_tokens = {"tokens": {"test-app": "secret-abc", "rotated": "fresh-secret"}} r = client.put("/configs/app_tokens", json=new_tokens, headers=self.HDRS) assert r.status_code == 200 - r = client.get("/configs/app_tokens", headers={"X-App-Token": "fresh-secret"}) - assert r.status_code == 200 - # the old token still works too (it was kept in the file) - r = client.get("/configs/app_tokens", headers=self.HDRS) - assert r.status_code == 200 \ No newline at end of file + # the store verifies against the replaced set immediately + assert tokens.verify("fresh-secret") + assert tokens.verify("secret-abc") + assert not tokens.verify("nope") \ No newline at end of file