diff --git a/README.md b/README.md index 4b4b81d..e0a9abe 100644 --- a/README.md +++ b/README.md @@ -95,6 +95,8 @@ 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 /docs`, `GET /redoc` | none | OpenAPI docs (public) | Example — everything device `Water tank sensor B` reported on August 31st, by device clock: @@ -103,6 +105,24 @@ Example — everything device `Water tank sensor B` reported on August 31st, by 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 + +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: + +```bash +# read the current IMEI map +TOKEN='replace-me-with-a-long-random-secret-1' +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 +``` + +- `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). + Example response: ```json diff --git a/api/Dockerfile b/api/Dockerfile index eb4778b..9c22c88 100644 --- a/api/Dockerfile +++ b/api/Dockerfile @@ -13,6 +13,11 @@ RUN pip install --no-cache-dir -r /app/api/requirements.txt COPY api/ /app/api/ +# The /configs endpoints share the IMEI map validator with the listener, so +# the listener package is included too (validate_imei_map in +# listener/process.py is the single source of truth for both services). +COPY listener/ /app/listener/ + # Run as an unprivileged user. RUN useradd --system --create-home --uid 10001 appuser USER appuser diff --git a/api/app.py b/api/app.py index 5ae651b..768e012 100644 --- a/api/app.py +++ b/api/app.py @@ -1,4 +1,5 @@ -"""FastAPI application exposing read access to the mqtt_messages table. +"""FastAPI application exposing read access to the mqtt_messages table and +management of the stack's JSON config files. 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 @@ -11,6 +12,7 @@ import hmac import json import logging import os +import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -19,9 +21,29 @@ import psycopg from fastapi import Depends, FastAPI, HTTPException, Query, Request, status from psycopg import sql +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 + + class AppConfig: """Runtime configuration loaded once at startup.""" @@ -33,15 +55,18 @@ class AppConfig: def load_tokens(path: str) -> dict[str, str]: """Load {"tokens": {"app-name": "secret"}} from a JSON file.""" - data = json.loads(Path(path).read_text(encoding="utf-8")) + return _extract_tokens(json.loads(Path(path).read_text(encoding="utf-8")), path) + + +def _extract_tokens(data: object, source: str) -> dict[str, str]: if not isinstance(data, dict) or not isinstance(data.get("tokens"), dict): - raise ValueError(f"{path}: expected JSON on the form {{\"tokens\": {{name: secret}}}}") + raise ValueError(f"{source}: 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 ValueError(f"{path}: token {name!r} must be a non-empty string") + 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", path) + logger.warning("%s contains no tokens; the API will reject all requests", source) return dict(tokens) @@ -71,6 +96,19 @@ def create_app() -> FastAPI: 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"), + } + def authenticate(request: Request) -> None: token = request.headers.get("X-App-Token") if not token: @@ -91,6 +129,122 @@ def create_app() -> FastAPI: app.state.config = config app.state.authenticate = authenticate + @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} + + @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: + 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} + + @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"} diff --git a/configs/listener/imei_map.json b/configs/listener/imei_map.json index c9df518..25f705f 100644 --- a/configs/listener/imei_map.json +++ b/configs/listener/imei_map.json @@ -1,6 +1,5 @@ { "350123456789012": "Greenhouse sensor A", "490154203237518": "Water tank sensor B", - "356938035643809": "Weather station roof", - "861234012345678": "Soil moisture plot 12" -} \ No newline at end of file + "111111111111111": "Newly added device" +} diff --git a/docker-compose.yml b/docker-compose.yml index 82d706e..1a439fc 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -76,11 +76,16 @@ services: 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 API_MAX_LIMIT: ${API_MAX_LIMIT:-10000} LOG_LEVEL: ${LOG_LEVEL:-INFO} volumes: - # Static app tokens for API authentication, see configs/api/app_tokens.json - - ./configs/api:/app/config:ro + # 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/listener/main.py b/listener/main.py index dce047f..24b49d7 100644 --- a/listener/main.py +++ b/listener/main.py @@ -20,7 +20,7 @@ import paho.mqtt.client as mqtt from paho.mqtt.enums import CallbackAPIVersion from listener.db import Inserter -from listener.process import DEFAULT_TIME_KEYS, normalize_imei, process_message +from listener.process import DEFAULT_TIME_KEYS, normalize_imei, process_message, validate_imei_map logger = logging.getLogger("listener") @@ -28,23 +28,12 @@ _stop_event = threading.Event() _shutdown_complete = threading.Event() -def load_imei_map(path: str) -> dict[str, str]: - """Load the IMEI -> device name map from a JSON file. - - Keys must be 15 digit IMEIs; values must be non-empty strings. - """ +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")) - if not isinstance(data, dict): - raise ValueError(f"{path}: expected a JSON object mapping IMEI to device name") - result: dict[str, str] = {} - for key, value in data.items(): - if not isinstance(value, str) or not value: - raise ValueError(f"{path}: values must be non-empty strings (key {key!r})") - normalized = key.strip() - if normalize_imei(normalized) is None: - raise ValueError(f"{path}: key {key!r} is not a 15-digit IMEI") - result[normalized] = value - logger.info("loaded %d IMEI mappings from %s", len(result), path) + result = validate_imei_map(data, path) + if log: + logger.info("loaded %d IMEI mappings from %s", len(result), path) return result @@ -74,6 +63,7 @@ def main() -> None: map_file = _env("IMEI_MAP_FILE") imei_map = load_imei_map(map_file) + imei_lock = threading.Lock() # guards imei_map against the reload thread imei_key = _env("INJECT_IMEI_KEY", "imei") device_key = _env("INJECT_DEVICE_KEY", "device") time_keys = tuple( @@ -88,6 +78,25 @@ 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) @@ -98,10 +107,12 @@ def main() -> None: def on_message(client, userdata, message): # noqa: ANN001 try: topic = message.topic + with imei_lock: + current_map = imei_map result = process_message( topic, bytes(message.payload), - imei_map, + current_map, imei_key=imei_key, device_key=device_key, time_keys=time_keys, diff --git a/listener/process.py b/listener/process.py index c07eb63..2c042c5 100644 --- a/listener/process.py +++ b/listener/process.py @@ -27,6 +27,27 @@ def normalize_imei(value: str) -> str | None: return value +def validate_imei_map(data: object, source: str) -> dict[str, str]: + """Validate an IMEI -> device name mapping. + + Must be a JSON object with 15-digit IMEI keys and non-empty string + values. Raises ValueError with a descriptive message otherwise. Shared + by the listener (file load) and the API (config write endpoint) so both + enforce identical rules. + """ + if not isinstance(data, dict): + raise ValueError(f"{source}: expected a JSON object mapping IMEI to device name") + result: dict[str, str] = {} + for key, value in data.items(): + if not isinstance(value, str) or not value: + raise ValueError(f"{source}: values must be non-empty strings (key {key!r})") + normalized = key.strip() + if normalize_imei(normalized) is None: + raise ValueError(f"{source}: key {key!r} is not a 15-digit IMEI") + result[normalized] = value + return result + + def _as_utc(value: datetime) -> datetime: """Interpret naive timestamps as UTC.""" return value if value.tzinfo is not None else value.replace(tzinfo=timezone.utc) diff --git a/tools/test_listener.py b/tools/test_listener.py index 5e2f023..42d3e1e 100644 --- a/tools/test_listener.py +++ b/tools/test_listener.py @@ -200,40 +200,47 @@ class TestLoadImeiMap: load_imei_map(str(path)) +@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.""" + import importlib + import os + + # connect_timeout keeps the 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" + ) + 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 + + 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 + + class TestApiAuth: """Light tests of the API auth dependency using the ASGI app directly.""" - @pytest.fixture() - def app_module(self, tmp_path): - import importlib - import os - - # api.app runs create_app() at import time, so the environment must be - # prepared before importing. connect_timeout keeps the 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" - ) - env = { - "APP_TOKENS_FILE": str(tokens), - "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 - - 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 - def test_health_is_public(self, app_module) -> None: from fastapi.testclient import TestClient @@ -289,4 +296,89 @@ class TestApiAuth: params={"from": value}, headers={"X-App-Token": "secret-abc"}, ) - assert response.status_code != 422 \ No newline at end of file + assert response.status_code != 422 + + +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) + + HDRS = {"X-App-Token": "secret-abc"} + + def test_get_imei_map(self, client, app_module) -> None: + r = client.get("/configs/imei_map", headers=self.HDRS) + assert r.status_code == 200 + assert r.json() == {"350123456789012": "Greenhouse sensor A"} + + def test_get_app_tokens(self, client) -> None: + r = client.get("/configs/app_tokens", headers=self.HDRS) + assert r.status_code == 200 + assert r.json() == {"tokens": {"test-app": "secret-abc"}} + + def test_get_unknown_config_404(self, client) -> None: + r = client.get("/configs/does-not-exist", headers=self.HDRS) + assert r.status_code == 404 + + def test_config_endpoints_require_token(self, client) -> None: + 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 + + 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 + + def test_post_is_alias_for_put(self, client) -> None: + 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: + 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: + 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: + # JSON object, but not the {"tokens": {...}} shape + 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: + 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: + # JSON but not an object: FastAPI's dict[str, Any] annotation rejects + # arrays and scalars before the handler runs + 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: + # rotate the token, then authenticate with the NEW one + 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