This commit is contained in:
Jonas Bengtsson
2026-08-31 22:39:20 +02:00
parent 26a027d1e8
commit ac52671b10
8 changed files with 367 additions and 60 deletions
+20
View File
@@ -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>": "<device name>"}` — 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
+5
View File
@@ -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
+159 -5
View File
@@ -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"}
+2 -3
View File
@@ -1,6 +1,5 @@
{
"350123456789012": "Greenhouse sensor A",
"490154203237518": "Water tank sensor B",
"356938035643809": "Weather station roof",
"861234012345678": "Soil moisture plot 12"
}
"111111111111111": "Newly added device"
}
+7 -2
View File
@@ -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:
+29 -18
View File
@@ -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,
+21
View File
@@ -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)
+124 -32
View File
@@ -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
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