Feature/ingestor management 481

README.md

@@ -85,6 +85,8 @@ The web app can be configured with environment variables (defaults shown):
 | `DEBUG` | `0` | Set to `1` for verbose logging in the web and ingestor services. |
 | `FEDERATION` | `1` | Set to `1` to announce your instance and crawl peers, or `0` to disable federation. Private mode overrides this. |
 | `PRIVATE` | `0` | Set to `1` to hide the chat UI, disable message APIs, and exclude hidden clients from public listings. |
+| `INGESTOR_MANAGEMENT` | `0` | Set to `1` to enable ingestor management features for community contributors. |
+| `ADMIN_TOKEN` | _unset_ | Token required to access ingestor management admin endpoints. |
 
 The application derives SEO-friendly document titles, descriptions, and social
 preview tags from these existing configuration values and reuses the bundled
@@ -144,6 +146,53 @@ The web app contains an API:
 
 The `API_TOKEN` environment variable must be set to a non-empty value and match the token supplied in the `Authorization` header for `POST` requests.
 
+### Ingestor Management
+
+When `INGESTOR_MANAGEMENT=1` is enabled, instance administrators can register
+external ingestors with individual API keys instead of sharing the main
+`API_TOKEN`. This allows community members to contribute data from their own
+Meshtastic nodes while maintaining control over access.
+
+Each ingestor registration tracks:
+
+* **API Key**: Unique authentication token for the ingestor
+* **Node ID**: Associated Meshtastic node identifier
+* **Version**: Ingestor software version (reported automatically)
+* **Last Request**: Timestamp of the most recent data submission
+* **Contact Info**: Email and/or Matrix username for the contributor
+
+#### Admin API Endpoints
+
+All admin endpoints require `ADMIN_TOKEN` in the `Authorization` header:
+
+| Method | Endpoint | Description |
+| --- | --- | --- |
+| `GET` | `/admin/ingestors` | List all registered ingestors |
+| `GET` | `/admin/ingestors/:id` | Get details for a specific ingestor |
+| `POST` | `/admin/ingestors` | Create a new ingestor registration |
+| `PATCH` | `/admin/ingestors/:id` | Update ingestor details |
+| `POST` | `/admin/ingestors/:id/regenerate-key` | Generate a new API key |
+| `POST` | `/admin/ingestors/:id/deactivate` | Disable an ingestor |
+| `POST` | `/admin/ingestors/:id/reactivate` | Re-enable an ingestor |
+| `DELETE` | `/admin/ingestors/:id` | Permanently remove an ingestor |
+
+#### Creating an Ingestor
+
+```bash
+curl -X POST https://your-instance.example/admin/ingestors \
+  -H "Authorization: Bearer YOUR_ADMIN_TOKEN" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "name": "Community Node Berlin",
+    "node_id": "!abc12345",
+    "contact_email": "[email protected]",
+    "contact_matrix": "@user:matrix.org"
+  }'
+```
+
+The response includes the generated API key. **Store it securely** - it will
+only be shown once (regeneration is possible but invalidates the previous key).
+
 ### Monitoring
 
 PotatoMesh ships with a Prometheus exporter mounted at `/metrics`. Consult

data/ingestors.sql

@@ -0,0 +1,33 @@
+-- Copyright © 2025-26 l5yth & contributors
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+PRAGMA journal_mode=WAL;
+
+CREATE TABLE IF NOT EXISTS ingestors (
+  id                TEXT PRIMARY KEY,
+  api_key           TEXT NOT NULL UNIQUE,
+  name              TEXT,
+  node_id           TEXT,
+  contact_email     TEXT,
+  contact_matrix    TEXT,
+  version           TEXT,
+  last_request_time INTEGER,
+  request_count     INTEGER DEFAULT 0,
+  created_at        INTEGER NOT NULL,
+  is_active         BOOLEAN NOT NULL DEFAULT 1
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_ingestors_api_key ON ingestors(api_key);
+CREATE INDEX IF NOT EXISTS idx_ingestors_node_id ON ingestors(node_id);
+CREATE INDEX IF NOT EXISTS idx_ingestors_is_active ON ingestors(is_active);

data/mesh_ingestor/queue.py

@@ -101,6 +101,7 @@ def _post_json(
     *,
     instance: str | None = None,
     api_token: str | None = None,
+    ingestor_version: str | None = None,
 ) -> None:
     """Send a JSON payload to the configured web API.
 
@@ -109,12 +110,20 @@ def _post_json(
         payload: JSON-serialisable body to transmit.
         instance: Optional override for :data:`config.INSTANCE`.
         api_token: Optional override for :data:`config.API_TOKEN`.
+        ingestor_version: Optional version string to report in headers.
     """
 
     if instance is None:
         instance = config.INSTANCE
     if api_token is None:
         api_token = config.API_TOKEN
+    if ingestor_version is None:
+        try:
+            from data import VERSION
+
+            ingestor_version = VERSION
+        except ImportError:
+            ingestor_version = None
 
     if not instance:
         return
@@ -132,6 +141,8 @@ def _post_json(
     }
     if api_token:
         headers["Authorization"] = f"Bearer {api_token}"
+    if ingestor_version:
+        headers["X-Ingestor-Version"] = ingestor_version
 
     req = urllib.request.Request(
         url,

tests/test_mesh.py

@@ -1482,6 +1482,68 @@ def read(self):
     assert mesh.json.loads(req.data.decode("utf-8")) == {"hello": "world"}
 
 
+def test_post_json_sends_ingestor_version_header(mesh_module, monkeypatch):
+    mesh = mesh_module
+    monkeypatch.setattr(mesh, "INSTANCE", "https://example.test")
+    monkeypatch.setattr(mesh, "API_TOKEN", "secret")
+
+    captured = {}
+
+    def fake_urlopen(req, timeout=0):
+        captured["req"] = req
+
+        class DummyResponse:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *exc):
+                return False
+
+            def read(self):
+                return b"ok"
+
+        return DummyResponse()
+
+    monkeypatch.setattr(mesh.urllib.request, "urlopen", fake_urlopen)
+
+    mesh._post_json("/api/test", {"hello": "world"}, ingestor_version="1.2.3")
+
+    req = captured["req"]
+    assert req.get_header("X-ingestor-version") == "1.2.3"
+
+
+def test_post_json_omits_ingestor_version_when_unavailable(mesh_module, monkeypatch):
+    mesh = mesh_module
+    monkeypatch.setattr(mesh, "INSTANCE", "https://example.test")
+    monkeypatch.setattr(mesh, "API_TOKEN", "secret")
+
+    captured = {}
+
+    def fake_urlopen(req, timeout=0):
+        captured["req"] = req
+
+        class DummyResponse:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *exc):
+                return False
+
+            def read(self):
+                return b"ok"
+
+        return DummyResponse()
+
+    monkeypatch.setattr(mesh.urllib.request, "urlopen", fake_urlopen)
+
+    # Explicitly pass empty string to skip version header
+    mesh._post_json("/api/test", {"data": "test"}, ingestor_version="")
+
+    req = captured["req"]
+    # X-Ingestor-Version header should not be present when version is empty
+    assert req.get_header("X-ingestor-version") is None
+
+
 def test_node_to_dict_handles_non_utf8_bytes(mesh_module):
     mesh = mesh_module
 

web/lib/potato_mesh/application.rb

@@ -52,8 +52,10 @@
 require_relative "application/data_processing"
 require_relative "application/filesystem"
 require_relative "application/instances"
+require_relative "application/ingestors"
 require_relative "application/routes/api"
 require_relative "application/routes/ingest"
+require_relative "application/routes/admin"
 require_relative "application/routes/root"
 
 module PotatoMesh
@@ -64,6 +66,7 @@ class Application < Sinatra::Base
     extend App::Identity
     extend App::Federation
     extend App::Instances
+    extend App::Ingestors
     extend App::Prometheus
     extend App::Queries
     extend App::DataProcessing
@@ -75,13 +78,15 @@ class Application < Sinatra::Base
     include App::Identity
     include App::Federation
     include App::Instances
+    include App::Ingestors
     include App::Prometheus
     include App::Queries
     include App::DataProcessing
     include App::Filesystem
 
     register App::Routes::Api
     register App::Routes::Ingest
+    register App::Routes::Admin
     register App::Routes::Root
 
     DEFAULT_PORT = 41_447

web/lib/potato_mesh/application/data_processing.rb

@@ -278,10 +278,37 @@ def upsert_node(db, node_id, n)
         end
       end
 
+      # Validate the request's authorization token.
+      #
+      # Accepts both the shared API_TOKEN and individual ingestor API keys.
+      # When an ingestor key is used, the request is recorded for tracking.
+      #
+      # @return [void]
       def require_token!
         token = ENV["API_TOKEN"]
         provided = request.env["HTTP_AUTHORIZATION"].to_s.sub(/^Bearer\s+/i, "")
-        halt 403, { error: "Forbidden" }.to_json unless token && !token.empty? && secure_token_match?(token, provided)
+
+        # First check against the shared API_TOKEN
+        if token && !token.empty? && secure_token_match?(token, provided)
+          return
+        end
+
+        # Then check against registered ingestor API keys
+        if PotatoMesh::Config.ingestor_management_enabled? && !provided.empty?
+          db = open_database
+          begin
+            ingestor = validate_ingestor_token(db, provided)
+            if ingestor
+              version = request.env["HTTP_X_INGESTOR_VERSION"]
+              record_ingestor_request(db, provided, version: version)
+              return
+            end
+          ensure
+            db&.close
+          end
+        end
+
+        halt 403, { error: "Forbidden" }.to_json
       end
 
       def secure_token_match?(expected, provided)

web/lib/potato_mesh/application/database.rb

@@ -99,7 +99,7 @@ def db_schema_present?
       def init_db
         FileUtils.mkdir_p(File.dirname(PotatoMesh::Config.db_path))
         db = open_database
-        %w[nodes messages positions telemetry neighbors instances traces].each do |schema|
+        %w[nodes messages positions telemetry neighbors instances traces ingestors].each do |schema|
           sql_file = File.expand_path("../../../../data/#{schema}.sql", __dir__)
           db.execute_batch(File.read(sql_file))
         end
@@ -192,6 +192,13 @@ def ensure_schema_upgrades
           traces_schema = File.expand_path("../../../../data/traces.sql", __dir__)
           db.execute_batch(File.read(traces_schema))
         end
+
+        ingestor_tables =
+          db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='ingestors'").flatten
+        if ingestor_tables.empty?
+          ingestors_schema = File.expand_path("../../../../data/ingestors.sql", __dir__)
+          db.execute_batch(File.read(ingestors_schema))
+        end
       rescue SQLite3::SQLException, Errno::ENOENT => e
         warn_log(
           "Failed to apply schema upgrade",

web/lib/potato_mesh/application/helpers.rb

@@ -421,6 +421,31 @@ def federation_enabled?
       def federation_announcements_active?
         federation_enabled? && !test_environment?
       end
+
+      # Validate the admin token from the Authorization header.
+      #
+      # @return [void]
+      def require_admin_token!
+        admin_token = PotatoMesh::Config.admin_token
+        provided = request.env["HTTP_AUTHORIZATION"].to_s.sub(/^Bearer\s+/i, "")
+
+        unless admin_token && !admin_token.empty? && secure_token_match?(admin_token, provided)
+          halt 403, { error: "Forbidden" }.to_json
+        end
+      end
+
+      # Mask an API key for display, showing only the first and last segments.
+      #
+      # @param api_key [String, nil] the full API key.
+      # @return [String, nil] masked key or nil.
+      def mask_api_key(api_key)
+        return nil if api_key.nil? || api_key.empty?
+
+        parts = api_key.split("-")
+        return "****" if parts.length < 2
+
+        "#{parts.first}-****-****-****-#{parts.last}"
+      end
     end
   end
 end