Handle encrypted messages (#173)

* Handle encrypted messages

* Remove redundant message node columns

* Preserve original numeric message senders

* Normalize message sender IDs in API responses

* Exclude encrypted messages from API responses

* run rufo

data/mesh.py

@@ -1001,8 +1001,11 @@ def store_packet_dict(p: dict):
         return
 
     text = _first(dec, "payload.text", "text", default=None)
-    if not text:
-        return  # ignore non-text packets
+    encrypted = _first(dec, "payload.encrypted", "encrypted", default=None)
+    if encrypted is None:
+        encrypted = _first(p, "encrypted", default=None)
+    if not text and not encrypted:
+        return  # ignore packets that lack text and encrypted payloads
 
     # port filter: only keep packets from the TEXT_MESSAGE_APP port
     if portnum and portnum not in {"1", "TEXT_MESSAGE_APP"}:
@@ -1046,6 +1049,7 @@ def store_packet_dict(p: dict):
         "channel": ch,
         "portnum": str(portnum) if portnum is not None else None,
         "text": text,
+        "encrypted": encrypted,
         "snr": float(snr) if snr is not None else None,
         "rssi": int(rssi) if rssi is not None else None,
         "hop_limit": int(hop) if hop is not None else None,

data/messages.sql

@@ -21,6 +21,7 @@ CREATE TABLE IF NOT EXISTS messages (
     channel   INTEGER,
     portnum   TEXT,
     text      TEXT,
+    encrypted TEXT,
     snr       REAL,
     rssi      INTEGER,
     hop_limit INTEGER

data/migrations/20250206_add_encrypted_message_columns.sql

@@ -0,0 +1,4 @@
+-- Add support for encrypted messages to the existing schema.
+BEGIN;
+ALTER TABLE messages ADD COLUMN encrypted TEXT;
+COMMIT;

tests/test_mesh.py

@@ -740,6 +740,7 @@ def test_store_packet_dict_uses_top_level_channel(mesh_module, monkeypatch):
     assert payload["channel"] == 5
     assert payload["portnum"] == "1"
     assert payload["text"] == "hi"
+    assert payload["encrypted"] is None
     assert payload["snr"] is None and payload["rssi"] is None
     assert priority == mesh._MESSAGE_POST_PRIORITY
 
@@ -770,6 +771,37 @@ def test_store_packet_dict_handles_invalid_channel(mesh_module, monkeypatch):
     path, payload, priority = captured[0]
     assert path == "/api/messages"
     assert payload["channel"] == 0
+    assert payload["encrypted"] is None
+    assert priority == mesh._MESSAGE_POST_PRIORITY
+
+
+def test_store_packet_dict_includes_encrypted_payload(mesh_module, monkeypatch):
+    mesh = mesh_module
+    captured = []
+    monkeypatch.setattr(
+        mesh,
+        "_queue_post_json",
+        lambda path, payload, *, priority: captured.append((path, payload, priority)),
+    )
+
+    packet = {
+        "id": 555,
+        "rxTime": 111,
+        "from": 2988082812,
+        "to": "!receiver",
+        "channel": 8,
+        "encrypted": "abc123==",
+    }
+
+    mesh.store_packet_dict(packet)
+
+    assert captured
+    path, payload, priority = captured[0]
+    assert path == "/api/messages"
+    assert payload["encrypted"] == "abc123=="
+    assert payload["text"] is None
+    assert payload["from_id"] == 2988082812
+    assert payload["to_id"] == "!receiver"
     assert priority == mesh._MESSAGE_POST_PRIORITY
 
 

web/app.rb

@@ -344,16 +344,17 @@ def query_messages(limit)
                       SELECT m.*, n.*, m.snr AS msg_snr
                       FROM messages m
                       LEFT JOIN nodes n ON (
-                        m.from_id = n.node_id OR (
-                          CAST(m.from_id AS TEXT) <> '' AND
-                          CAST(m.from_id AS TEXT) GLOB '[0-9]*' AND
-                          CAST(m.from_id AS INTEGER) = n.num
+                        m.from_id IS NOT NULL AND TRIM(m.from_id) <> '' AND (
+                          m.from_id = n.node_id OR (
+                            m.from_id GLOB '[0-9]*' AND CAST(m.from_id AS INTEGER) = n.num
+                          )
                         )
                       )
+                      WHERE COALESCE(TRIM(m.encrypted), '') = ''
                       ORDER BY m.rx_time DESC
                       LIMIT ?
                     SQL
-  msg_fields = %w[id rx_time rx_iso from_id to_id channel portnum text msg_snr rssi hop_limit]
+  msg_fields = %w[id rx_time rx_iso from_id to_id channel portnum text encrypted msg_snr rssi hop_limit]
   rows.each do |r|
     if DEBUG && (r["from_id"].nil? || r["from_id"].to_s.empty?)
       raw = db.execute("SELECT * FROM messages WHERE id = ?", [r["id"]]).first
@@ -366,7 +367,8 @@ def query_messages(limit)
       node[k] = r.delete(k)
     end
     r["snr"] = r.delete("msg_snr")
-    if r["from_id"] && (node["node_id"].nil? || node["node_id"].to_s.empty?)
+    references = [r["from_id"]].compact
+    if references.any? && (node["node_id"].nil? || node["node_id"].to_s.empty?)
       lookup_keys = []
       canonical = normalize_node_id(db, r["from_id"])
       lookup_keys << canonical if canonical
@@ -389,6 +391,16 @@ def query_messages(limit)
     end
     node["role"] = "CLIENT" if node.key?("role") && (node["role"].nil? || node["role"].to_s.empty?)
     r["node"] = node
+
+    canonical_from_id = string_or_nil(node["node_id"]) || string_or_nil(normalize_node_id(db, r["from_id"]))
+    if canonical_from_id
+      raw_from_id = string_or_nil(r["from_id"])
+      if raw_from_id.nil? || raw_from_id.match?(/\A[0-9]+\z/)
+        r["from_id"] = canonical_from_id
+      elsif raw_from_id.start_with?("!") && raw_from_id.casecmp(canonical_from_id) != 0
+        r["from_id"] = canonical_from_id
+      end
+    end
     if DEBUG && (r["from_id"].nil? || r["from_id"].to_s.empty?)
       Kernel.warn "[debug] row after processing: #{r.inspect}"
     end
@@ -886,54 +898,110 @@ end
 def insert_message(db, m)
   msg_id = m["id"] || m["packet_id"]
   return unless msg_id
+
   rx_time = m["rx_time"]&.to_i || Time.now.to_i
   rx_iso = m["rx_iso"] || Time.at(rx_time).utc.iso8601
+
   raw_from_id = m["from_id"]
   if raw_from_id.nil? || raw_from_id.to_s.strip.empty?
     alt_from = m["from"]
     raw_from_id = alt_from unless alt_from.nil? || alt_from.to_s.strip.empty?
   end
-  trimmed_from_id = raw_from_id.nil? ? nil : raw_from_id.to_s.strip
-  trimmed_from_id = nil if trimmed_from_id&.empty?
-  canonical_from_id = normalize_node_id(db, raw_from_id)
-  use_canonical = canonical_from_id && (trimmed_from_id.nil? || prefer_canonical_sender?(m))
-  from_id = if use_canonical
-      canonical_from_id.to_s.strip
-    else
-      trimmed_from_id
+
+  trimmed_from_id = string_or_nil(raw_from_id)
+  canonical_from_id = string_or_nil(normalize_node_id(db, raw_from_id))
+  from_id = trimmed_from_id
+  if canonical_from_id
+    if from_id.nil?
+      from_id = canonical_from_id
+    elsif prefer_canonical_sender?(m)
+      from_id = canonical_from_id
+    elsif from_id.start_with?("!") && from_id.casecmp(canonical_from_id) != 0
+      from_id = canonical_from_id
     end
-  from_id = nil if from_id&.empty?
+  end
+
+  raw_to_id = m["to_id"]
+  raw_to_id = m["to"] if raw_to_id.nil? || raw_to_id.to_s.strip.empty?
+  trimmed_to_id = string_or_nil(raw_to_id)
+  canonical_to_id = string_or_nil(normalize_node_id(db, raw_to_id))
+  to_id = trimmed_to_id
+  if canonical_to_id
+    if to_id.nil?
+      to_id = canonical_to_id
+    elsif to_id.start_with?("!") && to_id.casecmp(canonical_to_id) != 0
+      to_id = canonical_to_id
+    end
+  end
+
+  encrypted = string_or_nil(m["encrypted"])
+
   row = [
     msg_id,
     rx_time,
     rx_iso,
     from_id,
-    m["to_id"],
+    to_id,
     m["channel"],
     m["portnum"],
     m["text"],
+    encrypted,
     m["snr"],
     m["rssi"],
     m["hop_limit"],
   ]
+
   with_busy_retry do
-    existing = db.get_first_row("SELECT from_id FROM messages WHERE id = ?", [msg_id])
+    existing = db.get_first_row(
+      "SELECT from_id, to_id, encrypted FROM messages WHERE id = ?",
+      [msg_id],
+    )
     if existing
+      updates = {}
+
       if from_id
         existing_from = existing.is_a?(Hash) ? existing["from_id"] : existing[0]
         existing_from_str = existing_from&.to_s
         should_update = existing_from_str.nil? || existing_from_str.strip.empty?
         should_update ||= existing_from != from_id
-        db.execute("UPDATE messages SET from_id = ? WHERE id = ?", [from_id, msg_id]) if should_update
+        updates["from_id"] = from_id if should_update
+      end
+
+      if to_id
+        existing_to = existing.is_a?(Hash) ? existing["to_id"] : existing[1]
+        existing_to_str = existing_to&.to_s
+        should_update = existing_to_str.nil? || existing_to_str.strip.empty?
+        should_update ||= existing_to != to_id
+        updates["to_id"] = to_id if should_update
+      end
+
+      if encrypted
+        existing_encrypted = existing.is_a?(Hash) ? existing["encrypted"] : existing[2]
+        existing_encrypted_str = existing_encrypted&.to_s
+        should_update = existing_encrypted_str.nil? || existing_encrypted_str.strip.empty?
+        should_update ||= existing_encrypted != encrypted
+        updates["encrypted"] = encrypted if should_update
+      end
+
+      unless updates.empty?
+        assignments = updates.keys.map { |column| "#{column} = ?" }.join(", ")
+        db.execute("UPDATE messages SET #{assignments} WHERE id = ?", updates.values + [msg_id])
       end
     else
       begin
         db.execute <<~SQL, row
-                     INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,portnum,text,snr,rssi,hop_limit)
-                     VALUES (?,?,?,?,?,?,?,?,?,?,?)
+                     INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,portnum,text,encrypted,snr,rssi,hop_limit)
+                     VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
                    SQL
       rescue SQLite3::ConstraintException
-        db.execute("UPDATE messages SET from_id = ? WHERE id = ?", [from_id, msg_id]) if from_id
+        fallback_updates = {}
+        fallback_updates["from_id"] = from_id if from_id
+        fallback_updates["to_id"] = to_id if to_id
+        fallback_updates["encrypted"] = encrypted if encrypted
+        unless fallback_updates.empty?
+          assignments = fallback_updates.keys.map { |column| "#{column} = ?" }.join(", ")
+          db.execute("UPDATE messages SET #{assignments} WHERE id = ?", fallback_updates.values + [msg_id])
+        end
       end
     end
   end

web/spec/app_spec.rb

@@ -678,7 +678,9 @@ RSpec.describe "Potato Mesh Sinatra app" do
 
       with_db(readonly: true) do |db|
         db.results_as_hash = true
-        rows = db.execute("SELECT id, from_id, rx_time, rx_iso, text FROM messages ORDER BY id")
+        rows = db.execute(
+          "SELECT id, from_id, to_id, rx_time, rx_iso, text, encrypted FROM messages ORDER BY id",
+        )
 
         expect(rows.size).to eq(2)
 
@@ -686,18 +688,116 @@ RSpec.describe "Potato Mesh Sinatra app" do
 
         expect(first["id"]).to eq(101)
         expect(first["from_id"]).to eq(node_id)
+        expect(first).not_to have_key("from_node_id")
+        expect(first).not_to have_key("from_node_num")
         expect(first["rx_time"]).to eq(reference_time.to_i)
         expect(first["rx_iso"]).to eq(reference_time.utc.iso8601)
         expect(first["text"]).to eq("normalized")
+        expect(first).not_to have_key("to_node_id")
+        expect(first).not_to have_key("to_node_num")
+        expect(first["encrypted"]).to be_nil
 
         expect(second["id"]).to eq(102)
         expect(second["from_id"]).to be_nil
+        expect(second).not_to have_key("from_node_id")
+        expect(second).not_to have_key("from_node_num")
         expect(second["rx_time"]).to eq(reference_time.to_i)
         expect(second["rx_iso"]).to eq(reference_time.utc.iso8601)
         expect(second["text"]).to eq("blank")
+        expect(second).not_to have_key("to_node_id")
+        expect(second).not_to have_key("to_node_num")
+        expect(second["encrypted"]).to be_nil
       end
     end
 
+    it "stores encrypted messages and resolves node references" do
+      sender_id = "!feedc0de"
+      sender_num = 0xfeedc0de
+      receiver_id = "!c0ffee99"
+      receiver_num = 0xc0ffee99
+
+      sender_node = {
+        "node_id" => sender_id,
+        "short_name" => "EncS",
+        "long_name" => "Encrypted Sender",
+        "hw_model" => "TEST",
+        "role" => "CLIENT",
+        "snr" => 5.5,
+        "battery_level" => 80.0,
+        "voltage" => 3.9,
+        "last_heard" => reference_time.to_i - 30,
+        "position_time" => reference_time.to_i - 60,
+        "latitude" => 52.1,
+        "longitude" => 13.1,
+        "altitude" => 42.0,
+      }
+      sender_payload = build_node_payload(sender_node)
+      sender_payload["num"] = sender_num
+
+      receiver_node = {
+        "node_id" => receiver_id,
+        "short_name" => "EncR",
+        "long_name" => "Encrypted Receiver",
+        "hw_model" => "TEST",
+        "role" => "CLIENT",
+        "snr" => 4.25,
+        "battery_level" => 75.0,
+        "voltage" => 3.8,
+        "last_heard" => reference_time.to_i - 40,
+        "position_time" => reference_time.to_i - 70,
+        "latitude" => 52.2,
+        "longitude" => 13.2,
+        "altitude" => 35.0,
+      }
+      receiver_payload = build_node_payload(receiver_node)
+      receiver_payload["num"] = receiver_num
+
+      post "/api/nodes", { sender_id => sender_payload }.to_json, auth_headers
+      expect(last_response).to be_ok
+      post "/api/nodes", { receiver_id => receiver_payload }.to_json, auth_headers
+      expect(last_response).to be_ok
+
+      encrypted_b64 = Base64.strict_encode64("secret message")
+      payload = {
+        "packet_id" => 777_001,
+        "rx_time" => reference_time.to_i,
+        "rx_iso" => reference_time.utc.iso8601,
+        "from_id" => sender_num.to_s,
+        "to_id" => receiver_id,
+        "channel" => 8,
+        "portnum" => "TEXT_MESSAGE_APP",
+        "encrypted" => encrypted_b64,
+        "snr" => -12.5,
+        "rssi" => -109,
+        "hop_limit" => 3,
+      }
+
+      post "/api/messages", payload.to_json, auth_headers
+
+      expect(last_response).to be_ok
+      expect(JSON.parse(last_response.body)).to eq("status" => "ok")
+
+      with_db(readonly: true) do |db|
+        db.results_as_hash = true
+        row = db.get_first_row(
+          "SELECT from_id, to_id, text, encrypted FROM messages WHERE id = ?",
+          [777_001],
+        )
+
+        expect(row["from_id"]).to eq(sender_id)
+        expect(row["to_id"]).to eq(receiver_id)
+        expect(row["text"]).to be_nil
+        expect(row["encrypted"]).to eq(encrypted_b64)
+      end
+
+      get "/api/messages"
+      expect(last_response).to be_ok
+
+      messages = JSON.parse(last_response.body)
+      expect(messages).to be_an(Array)
+      expect(messages).to be_empty
+    end
+
     it "stores messages containing SQL control characters without executing them" do
       payload = {
         "packet_id" => 404,
@@ -901,11 +1001,28 @@ RSpec.describe "Potato Mesh Sinatra app" do
 
         expect(actual_row["rx_time"]).to eq(expected["rx_time"])
         expect(actual_row["rx_iso"]).to eq(expected["rx_iso"])
-        expect(actual_row["from_id"]).to eq(expected["from_id"])
-        expect(actual_row["to_id"]).to eq(expected["to_id"])
+
+        expected_from_id = expected["from_id"]
+        if expected_from_id.is_a?(String) && expected_from_id.match?(/\A[0-9]+\z/)
+          expected_from_id = node_aliases[expected_from_id] || expected_from_id
+        elsif expected_from_id.nil?
+          expected_from_id = message.dig("node", "node_id")
+        end
+        expect(actual_row["from_id"]).to eq(expected_from_id)
+        expect(actual_row).not_to have_key("from_node_id")
+        expect(actual_row).not_to have_key("from_node_num")
+
+        expected_to_id = expected["to_id"]
+        if expected_to_id.is_a?(String) && expected_to_id.match?(/\A[0-9]+\z/)
+          expected_to_id = node_aliases[expected_to_id] || expected_to_id
+        end
+        expect(actual_row["to_id"]).to eq(expected_to_id)
+        expect(actual_row).not_to have_key("to_node_id")
+        expect(actual_row).not_to have_key("to_node_num")
         expect(actual_row["channel"]).to eq(expected["channel"])
         expect(actual_row["portnum"]).to eq(expected["portnum"])
         expect(actual_row["text"]).to eq(expected["text"])
+        expect(actual_row["encrypted"]).to eq(expected["encrypted"])
         expect_same_value(actual_row["snr"], expected["snr"])
         expect(actual_row["rssi"]).to eq(expected["rssi"])
         expect(actual_row["hop_limit"]).to eq(expected["hop_limit"])

web/views/index.erb

@@ -1092,6 +1092,7 @@
         entries.push({ type: 'node', ts: n.first_heard ?? 0, item: n });
       }
       for (const m of messages || []) {
+        if (!m || m.encrypted) continue;
         entries.push({ type: 'msg', ts: m.rx_time ?? 0, item: m });
       }
       const nowSeconds = Math.floor(Date.now() / 1000);