Merge pull request #116 from kizniche/feat-int-mc-map-auto-uploader

Add automatic mesh map upload (integration/fanout module). Closes #108. Thank you!!

app/fanout/AGENTS_fanout.md

@@ -89,6 +89,19 @@ Amazon SQS delivery. Config blob:
 - Publishes a JSON envelope of the form `{"event_type":"message"|"raw_packet","data":...}`
 - Supports both decoded messages and raw packets via normal scope selection
 
+### map_upload (map_upload.py)
+Uploads heard repeater and room-server advertisements to map.meshcore.dev. Config blob:
+- `api_url` (optional, default `""`) — upload endpoint; empty falls back to the public map.meshcore.dev API
+- `dry_run` (bool, default `true`) — when true, logs the payload at INFO level without sending
+- `geofence_enabled` (bool, default `false`) — when true, only uploads nodes within `geofence_radius_km` of the radio's own configured lat/lon
+- `geofence_radius_km` (float, default `0`) — filter radius in kilometres
+
+Geofence notes:
+- The reference center is always the radio's own `adv_lat`/`adv_lon` from `radio_runtime.meshcore.self_info`, read **live at upload time** — no lat/lon is stored in the fanout config itself.
+- If the radio's lat/lon is `(0, 0)` or the radio is not connected, the geofence check is silently skipped so uploads continue normally until coordinates are configured.
+- Requires the radio to have `ENABLE_PRIVATE_KEY_EXPORT=1` firmware to sign uploads.
+- Scope is always `{"messages": "none", "raw_packets": "all"}` — only raw RF packets are processed.
+
 ## Adding a New Integration Type
 
 ### Step-by-step checklist
@@ -291,6 +304,7 @@ Migrations:
 - `app/fanout/webhook.py` — Webhook fanout module
 - `app/fanout/apprise_mod.py` — Apprise fanout module
 - `app/fanout/sqs.py` — Amazon SQS fanout module
+- `app/fanout/map_upload.py` — Map Upload fanout module
 - `app/repository/fanout.py` — Database CRUD
 - `app/routers/fanout.py` — REST API
 - `app/websocket.py` — `broadcast_event()` dispatches to fanout

app/fanout/community_mqtt.py

@@ -20,9 +20,9 @@ from datetime import datetime
 from typing import Any, Protocol
 
 import aiomqtt
-import nacl.bindings
 
 from app.fanout.mqtt_base import BaseMqttPublisher
+from app.keystore import ed25519_sign_expanded
 from app.path_utils import parse_packet_envelope, split_path_hex
 from app.version_info import get_app_build_info
 
@@ -40,9 +40,6 @@ _TOKEN_RENEWAL_THRESHOLD = _TOKEN_LIFETIME - 3600  # 23 hours
 _STATS_REFRESH_INTERVAL = 300  # 5 minutes
 _STATS_MIN_CACHE_SECS = 60  # Don't re-fetch stats within 60s
 
-# Ed25519 group order
-_L = 2**252 + 27742317777372353535851937790883648493
-
 # Route type mapping: bottom 2 bits of first byte
 _ROUTE_MAP = {0: "F", 1: "F", 2: "D", 3: "T"}
 
@@ -69,28 +66,6 @@ def _base64url_encode(data: bytes) -> str:
     return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
 
 
-def _ed25519_sign_expanded(
-    message: bytes, scalar: bytes, prefix: bytes, public_key: bytes
-) -> bytes:
-    """Sign a message using MeshCore's expanded Ed25519 key format.
-
-    MeshCore stores 64-byte "orlp" format keys: scalar(32) || prefix(32).
-    Standard Ed25519 libraries expect seed format and would re-SHA-512 the key.
-    This performs the signing manually using the already-expanded key material.
-
-    Port of meshcore-packet-capture's ed25519_sign_with_expanded_key().
-    """
-    # r = SHA-512(prefix || message) mod L
-    r = int.from_bytes(hashlib.sha512(prefix + message).digest(), "little") % _L
-    # R = r * B (base point multiplication)
-    R = nacl.bindings.crypto_scalarmult_ed25519_base_noclamp(r.to_bytes(32, "little"))
-    # k = SHA-512(R || public_key || message) mod L
-    k = int.from_bytes(hashlib.sha512(R + public_key + message).digest(), "little") % _L
-    # s = (r + k * scalar) mod L
-    s = (r + k * int.from_bytes(scalar, "little")) % _L
-    return R + s.to_bytes(32, "little")
-
-
 def _generate_jwt_token(
     private_key: bytes,
     public_key: bytes,
@@ -127,7 +102,7 @@ def _generate_jwt_token(
 
     scalar = private_key[:32]
     prefix = private_key[32:]
-    signature = _ed25519_sign_expanded(signing_input, scalar, prefix, public_key)
+    signature = ed25519_sign_expanded(signing_input, scalar, prefix, public_key)
 
     return f"{header_b64}.{payload_b64}.{signature.hex()}"
 

app/fanout/manager.py

@@ -21,6 +21,7 @@ def _register_module_types() -> None:
         return
     from app.fanout.apprise_mod import AppriseModule
     from app.fanout.bot import BotModule
+    from app.fanout.map_upload import MapUploadModule
     from app.fanout.mqtt_community import MqttCommunityModule
     from app.fanout.mqtt_private import MqttPrivateModule
     from app.fanout.sqs import SqsModule
@@ -32,6 +33,7 @@ def _register_module_types() -> None:
     _MODULE_TYPES["webhook"] = WebhookModule
     _MODULE_TYPES["apprise"] = AppriseModule
     _MODULE_TYPES["sqs"] = SqsModule
+    _MODULE_TYPES["map_upload"] = MapUploadModule
 
 
 def _matches_filter(filter_value: Any, key: str) -> bool:

app/fanout/map_upload.py

@@ -0,0 +1,320 @@
+"""Fanout module for uploading heard advert packets to map.meshcore.dev.
+
+Mirrors the logic of the standalone map.meshcore.dev-uploader project:
+- Listens on raw RF packets via on_raw
+- Filters for ADVERT packets, only processes repeaters (role 2) and rooms (role 3)
+- Skips nodes with no valid location (lat/lon None)
+- Applies per-pubkey rate-limiting (1-hour window, matching the uploader)
+- Signs the upload request with the radio's own Ed25519 private key
+- POSTs to the map API (or logs in dry-run mode)
+
+Dry-run mode (default: True) logs the full would-be payload at INFO level
+without making any HTTP requests. Disable it only after verifying the log
+output looks correct — in particular the radio params (freq/bw/sf/cr) and
+the raw hex link.
+
+Config keys
+-----------
+api_url : str, default ""
+    Upload endpoint. Empty string falls back to the public map.meshcore.dev API.
+dry_run : bool, default True
+    When True, log the payload at INFO level instead of sending it.
+geofence_enabled : bool, default False
+    When True, only upload nodes whose location falls within geofence_radius_km of
+    the radio's own configured latitude/longitude (read live from the radio at upload
+    time — no lat/lon is stored in this config).  When the radio's lat/lon is not set
+    (0, 0) or unavailable, the geofence check is silently skipped so uploads continue
+    normally until coordinates are configured.
+geofence_radius_km : float, default 0.0
+    Radius of the geofence in kilometres. Nodes further than this distance
+    from the radio's own position are skipped.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import math
+
+import httpx
+
+from app.decoder import parse_advertisement, parse_packet
+from app.fanout.base import FanoutModule
+from app.keystore import ed25519_sign_expanded, get_private_key, get_public_key
+from app.services.radio_runtime import radio_runtime
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_API_URL = "https://map.meshcore.dev/api/v1/uploader/node"
+
+# Re-upload guard: skip re-uploading a pubkey seen within this window (AU parity)
+_REUPLOAD_SECONDS = 3600
+
+# Only upload repeaters (2) and rooms (3). Any other role — including future
+# roles not yet defined — is rejected. An allowlist is used rather than a
+# blocklist so that new roles cannot accidentally start populating the map.
+_ALLOWED_DEVICE_ROLES = {2, 3}
+
+
+def _get_radio_params() -> dict:
+    """Read radio frequency parameters from the connected radio's self_info.
+
+    The Python meshcore library returns radio_freq in MHz (e.g. 910.525) and
+    radio_bw in kHz (e.g. 62.5). These are exactly the units the map API
+    expects, matching what the JS reference uploader produces after its own
+    /1000 division on raw integer values. No further scaling is applied here.
+    """
+    try:
+        mc = radio_runtime.meshcore
+        if not mc:
+            return {"freq": 0, "cr": 0, "sf": 0, "bw": 0}
+        info = mc.self_info
+        if not isinstance(info, dict):
+            return {"freq": 0, "cr": 0, "sf": 0, "bw": 0}
+        freq = info.get("radio_freq", 0) or 0
+        bw = info.get("radio_bw", 0) or 0
+        sf = info.get("radio_sf", 0) or 0
+        cr = info.get("radio_cr", 0) or 0
+        return {
+            "freq": freq,
+            "cr": cr,
+            "sf": sf,
+            "bw": bw,
+        }
+    except Exception as exc:
+        logger.debug("MapUpload: could not read radio params: %s", exc)
+    return {"freq": 0, "cr": 0, "sf": 0, "bw": 0}
+
+
+_ROLE_NAMES: dict[int, str] = {2: "repeater", 3: "room"}
+
+
+def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    """Return the great-circle distance in kilometres between two lat/lon points."""
+    r = 6371.0
+    phi1, phi2 = math.radians(lat1), math.radians(lat2)
+    dphi = math.radians(lat2 - lat1)
+    dlam = math.radians(lon2 - lon1)
+    a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlam / 2) ** 2
+    return 2 * r * math.asin(math.sqrt(a))
+
+
+class MapUploadModule(FanoutModule):
+    """Uploads heard ADVERT packets to the MeshCore community map."""
+
+    def __init__(self, config_id: str, config: dict, *, name: str = "") -> None:
+        super().__init__(config_id, config, name=name)
+        self._client: httpx.AsyncClient | None = None
+        self._last_error: str | None = None
+        # Per-pubkey rate limiting: pubkey_hex -> last_uploaded_advert_timestamp
+        self._seen: dict[str, int] = {}
+
+    async def start(self) -> None:
+        self._client = httpx.AsyncClient(timeout=httpx.Timeout(15.0))
+        self._last_error = None
+        self._seen.clear()
+
+    async def stop(self) -> None:
+        if self._client:
+            await self._client.aclose()
+            self._client = None
+        self._last_error = None
+
+    async def on_raw(self, data: dict) -> None:
+        if data.get("payload_type") != "ADVERT":
+            return
+
+        raw_hex = data.get("data", "")
+        if not raw_hex:
+            return
+
+        try:
+            raw_bytes = bytes.fromhex(raw_hex)
+        except ValueError:
+            return
+
+        packet_info = parse_packet(raw_bytes)
+        if packet_info is None:
+            return
+
+        advert = parse_advertisement(packet_info.payload, raw_packet=raw_bytes)
+        if advert is None:
+            return
+
+        # TODO: advert Ed25519 signature verification is skipped here.
+        # The radio has already validated the packet before passing it to RT,
+        # so re-verification is redundant in practice. If added, verify that
+        # nacl.bindings.crypto_sign_open(sig + (pubkey_bytes || timestamp_bytes),
+        # advert.public_key_bytes) succeeds before proceeding.
+
+        # Only process repeaters (2) and rooms (3) — any other role is rejected
+        if advert.device_role not in _ALLOWED_DEVICE_ROLES:
+            return
+
+        # Skip nodes with no valid location — the decoder already nulls out
+        # impossible values, so None means either no location flag or bad coords.
+        if advert.lat is None or advert.lon is None:
+            logger.debug(
+                "MapUpload: skipping %s — no valid location",
+                advert.public_key[:12],
+            )
+            return
+
+        pubkey = advert.public_key.lower()
+
+        # Rate-limit: skip if this pubkey's timestamp hasn't advanced enough
+        last_seen = self._seen.get(pubkey)
+        if last_seen is not None:
+            if last_seen >= advert.timestamp:
+                logger.debug(
+                    "MapUpload: skipping %s — possible replay (last=%d, advert=%d)",
+                    pubkey[:12],
+                    last_seen,
+                    advert.timestamp,
+                )
+                return
+            if advert.timestamp < last_seen + _REUPLOAD_SECONDS:
+                logger.debug(
+                    "MapUpload: skipping %s — within 1-hr rate-limit window (delta=%ds)",
+                    pubkey[:12],
+                    advert.timestamp - last_seen,
+                )
+                return
+
+        await self._upload(
+            pubkey, advert.timestamp, advert.device_role, raw_hex, advert.lat, advert.lon
+        )
+
+    async def _upload(
+        self,
+        pubkey: str,
+        advert_timestamp: int,
+        device_role: int,
+        raw_hex: str,
+        lat: float,
+        lon: float,
+    ) -> None:
+        # Geofence check: if enabled, skip nodes outside the configured radius.
+        # The reference center is the radio's own lat/lon read live from self_info —
+        # no coordinates are stored in the fanout config. If the radio lat/lon is
+        # (0, 0) or unavailable the check is skipped transparently so uploads
+        # continue normally until the operator sets coordinates in radio settings.
+        geofence_dist_km: float | None = None
+        if self.config.get("geofence_enabled"):
+            try:
+                mc = radio_runtime.meshcore
+                sinfo = mc.self_info if mc else None
+                fence_lat = float((sinfo or {}).get("adv_lat", 0) or 0)
+                fence_lon = float((sinfo or {}).get("adv_lon", 0) or 0)
+            except Exception as exc:
+                logger.debug("MapUpload: could not read radio lat/lon for geofence: %s", exc)
+                fence_lat = 0.0
+                fence_lon = 0.0
+
+            if fence_lat == 0.0 and fence_lon == 0.0:
+                logger.debug(
+                    "MapUpload: geofence skipped for %s — radio lat/lon not configured",
+                    pubkey[:12],
+                )
+            else:
+                fence_radius_km = float(self.config.get("geofence_radius_km", 0) or 0)
+                geofence_dist_km = _haversine_km(fence_lat, fence_lon, lat, lon)
+                if geofence_dist_km > fence_radius_km:
+                    logger.debug(
+                        "MapUpload: skipping %s — outside geofence (%.2f km > %.2f km)",
+                        pubkey[:12],
+                        geofence_dist_km,
+                        fence_radius_km,
+                    )
+                    return
+
+        private_key = get_private_key()
+        public_key = get_public_key()
+
+        if private_key is None or public_key is None:
+            logger.warning(
+                "MapUpload: private key not available — cannot sign upload for %s. "
+                "Ensure radio firmware has ENABLE_PRIVATE_KEY_EXPORT=1.",
+                pubkey[:12],
+            )
+            return
+
+        api_url = str(self.config.get("api_url", "") or _DEFAULT_API_URL).strip()
+        dry_run = bool(self.config.get("dry_run", True))
+        role_name = _ROLE_NAMES.get(device_role, f"role={device_role}")
+
+        params = _get_radio_params()
+        upload_data = {
+            "params": params,
+            "links": [f"meshcore://{raw_hex}"],
+        }
+
+        # Sign: SHA-256 the compact JSON, then Ed25519-sign the hash
+        json_str = json.dumps(upload_data, separators=(",", ":"))
+        data_hash = hashlib.sha256(json_str.encode()).digest()
+        scalar = private_key[:32]
+        prefix_bytes = private_key[32:]
+        signature = ed25519_sign_expanded(data_hash, scalar, prefix_bytes, public_key)
+
+        request_payload = {
+            "data": json_str,
+            "signature": signature.hex(),
+            "publicKey": public_key.hex(),
+        }
+
+        if dry_run:
+            geofence_note = (
+                f" | geofence: {geofence_dist_km:.2f} km from observer"
+                if geofence_dist_km is not None
+                else ""
+            )
+            logger.info(
+                "MapUpload [DRY RUN] %s (%s)%s → would POST to %s\n  payload: %s",
+                pubkey[:12],
+                role_name,
+                geofence_note,
+                api_url,
+                json.dumps(request_payload, separators=(",", ":")),
+            )
+            # Still update _seen so rate-limiting works during dry-run testing
+            self._seen[pubkey] = advert_timestamp
+            return
+
+        if not self._client:
+            return
+
+        try:
+            resp = await self._client.post(
+                api_url,
+                content=json.dumps(request_payload, separators=(",", ":")),
+                headers={"Content-Type": "application/json"},
+            )
+            resp.raise_for_status()
+            self._seen[pubkey] = advert_timestamp
+            self._last_error = None
+            logger.info(
+                "MapUpload: uploaded %s (%s) → HTTP %d",
+                pubkey[:12],
+                role_name,
+                resp.status_code,
+            )
+        except httpx.HTTPStatusError as exc:
+            self._last_error = f"HTTP {exc.response.status_code}"
+            logger.warning(
+                "MapUpload: server returned %d for %s: %s",
+                exc.response.status_code,
+                pubkey[:12],
+                exc.response.text[:200],
+            )
+        except httpx.RequestError as exc:
+            self._last_error = str(exc)
+            logger.warning("MapUpload: request error for %s: %s", pubkey[:12], exc)
+
+    @property
+    def status(self) -> str:
+        if self._client is None:
+            return "disconnected"
+        if self._last_error:
+            return "error"
+        return "connected"

app/keystore.py

@@ -1,14 +1,18 @@
 """
-Ephemeral keystore for storing sensitive keys in memory.
+Ephemeral keystore for storing sensitive keys in memory, plus the Ed25519
+signing primitive used by fanout modules that need to sign requests with the
+radio's own key.
 
 The private key is stored in memory only and is never persisted to disk.
 It's exported from the radio on startup and reconnect, then used for
 server-side decryption of direct messages.
 """
 
+import hashlib
 import logging
 from typing import TYPE_CHECKING
 
+import nacl.bindings
 from meshcore import EventType
 
 from app.decoder import derive_public_key
@@ -25,11 +29,30 @@ NO_EVENT_RECEIVED_GUIDANCE = (
     "issue commands to the radio."
 )
 
+# Ed25519 group order (L) — used in the expanded signing primitive below
+_L = 2**252 + 27742317777372353535851937790883648493
+
 # In-memory storage for the private key and derived public key
 _private_key: bytes | None = None
 _public_key: bytes | None = None
 
 
+def ed25519_sign_expanded(message: bytes, scalar: bytes, prefix: bytes, public_key: bytes) -> bytes:
+    """Sign a message using MeshCore's expanded Ed25519 key format.
+
+    MeshCore stores 64-byte keys as scalar(32) || prefix(32).  Standard
+    Ed25519 libraries expect seed format and would re-SHA-512 the key, so we
+    perform the signing manually using the already-expanded key material.
+
+    Port of meshcore-packet-capture's ed25519_sign_with_expanded_key().
+    """
+    r = int.from_bytes(hashlib.sha512(prefix + message).digest(), "little") % _L
+    R = nacl.bindings.crypto_scalarmult_ed25519_base_noclamp(r.to_bytes(32, "little"))
+    k = int.from_bytes(hashlib.sha512(R + public_key + message).digest(), "little") % _L
+    s = (r + k * int.from_bytes(scalar, "little")) % _L
+    return R + s.to_bytes(32, "little")
+
+
 def clear_keys() -> None:
     """Clear any stored private/public key material from memory."""
     global _private_key, _public_key

app/routers/fanout.py

@@ -16,7 +16,7 @@ from app.repository.fanout import FanoutConfigRepository
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/fanout", tags=["fanout"])
 
-_VALID_TYPES = {"mqtt_private", "mqtt_community", "bot", "webhook", "apprise", "sqs"}
+_VALID_TYPES = {"mqtt_private", "mqtt_community", "bot", "webhook", "apprise", "sqs", "map_upload"}
 
 _IATA_RE = re.compile(r"^[A-Z]{3}$")
 _DEFAULT_COMMUNITY_MQTT_TOPIC_TEMPLATE = "meshcore/{IATA}/{PUBLIC_KEY}/packets"
@@ -94,6 +94,8 @@ def _validate_and_normalize_config(config_type: str, config: dict) -> dict:
         _validate_apprise_config(normalized)
     elif config_type == "sqs":
         _validate_sqs_config(normalized)
+    elif config_type == "map_upload":
+        _validate_map_upload_config(normalized)
 
     return normalized
 
@@ -295,10 +297,33 @@ def _validate_sqs_config(config: dict) -> None:
         )
 
 
+def _validate_map_upload_config(config: dict) -> None:
+    """Validate and normalize map_upload config blob."""
+    api_url = str(config.get("api_url", "")).strip()
+    if api_url and not api_url.startswith(("http://", "https://")):
+        raise HTTPException(
+            status_code=400,
+            detail="api_url must start with http:// or https://",
+        )
+    # Persist the cleaned value (empty string means use the module default)
+    config["api_url"] = api_url
+    config["dry_run"] = bool(config.get("dry_run", True))
+    config["geofence_enabled"] = bool(config.get("geofence_enabled", False))
+    try:
+        radius = float(config.get("geofence_radius_km", 0) or 0)
+    except (TypeError, ValueError):
+        raise HTTPException(status_code=400, detail="geofence_radius_km must be a number") from None
+    if radius < 0:
+        raise HTTPException(status_code=400, detail="geofence_radius_km must be >= 0")
+    config["geofence_radius_km"] = radius
+
+
 def _enforce_scope(config_type: str, scope: dict) -> dict:
     """Enforce type-specific scope constraints. Returns normalized scope."""
     if config_type == "mqtt_community":
         return {"messages": "none", "raw_packets": "all"}
+    if config_type == "map_upload":
+        return {"messages": "none", "raw_packets": "all"}
     if config_type == "bot":
         return {"messages": "all", "raw_packets": "none"}
     if config_type in ("webhook", "apprise"):