diff --git a/app/database.py b/app/database.py index a3ec77a..6a4d653 100644 --- a/app/database.py +++ b/app/database.py @@ -482,6 +482,105 @@ class Database: cursor = conn.execute("DELETE FROM channels WHERE idx = ?", (idx,)) return cursor.rowcount > 0 + # ================================================================ + # Regions (MeshCore flood scopes) + # ================================================================ + + def create_region(self, name: str, key_hex: str) -> int: + """Insert a new region. Raises sqlite3.IntegrityError on duplicate name.""" + with self._connect() as conn: + cursor = conn.execute( + """INSERT INTO regions (name, key_hex) VALUES (?, ?)""", + (name, key_hex) + ) + return cursor.lastrowid + + def list_regions(self) -> List[Dict]: + with self._connect() as conn: + rows = conn.execute( + "SELECT * FROM regions ORDER BY name COLLATE NOCASE" + ).fetchall() + return [dict(r) for r in rows] + + def get_region(self, region_id: int) -> Optional[Dict]: + with self._connect() as conn: + row = conn.execute( + "SELECT * FROM regions WHERE id = ?", (region_id,) + ).fetchone() + return dict(row) if row else None + + def get_region_by_name(self, name: str) -> Optional[Dict]: + with self._connect() as conn: + row = conn.execute( + "SELECT * FROM regions WHERE name = ?", (name,) + ).fetchone() + return dict(row) if row else None + + def delete_region(self, region_id: int) -> bool: + with self._connect() as conn: + cursor = conn.execute("DELETE FROM regions WHERE id = ?", (region_id,)) + return cursor.rowcount > 0 + + def set_default_region(self, region_id: Optional[int]) -> None: + """Clear any existing default, then set the given region as default. + + Passing None clears the default flag on all regions. + """ + with self._connect() as conn: + conn.execute("UPDATE regions SET is_default = 0, updated_at = datetime('now') WHERE is_default = 1") + if region_id is not None: + conn.execute( + "UPDATE regions SET is_default = 1, updated_at = datetime('now') WHERE id = ?", + (region_id,) + ) + + def get_default_region(self) -> Optional[Dict]: + with self._connect() as conn: + row = conn.execute( + "SELECT * FROM regions WHERE is_default = 1 LIMIT 1" + ).fetchone() + return dict(row) if row else None + + def set_channel_scope(self, channel_idx: int, region_id: Optional[int]) -> None: + """Set or clear the region mapping for a channel. + + region_id=None removes the mapping (firmware default will apply). + """ + with self._connect() as conn: + if region_id is None: + conn.execute("DELETE FROM channel_scopes WHERE channel_idx = ?", (channel_idx,)) + else: + conn.execute( + """INSERT INTO channel_scopes (channel_idx, region_id) + VALUES (?, ?) + ON CONFLICT(channel_idx) DO UPDATE SET + region_id = excluded.region_id, + updated_at = datetime('now')""", + (channel_idx, region_id) + ) + + def get_channel_scope(self, channel_idx: int) -> Optional[Dict]: + """Return the region dict assigned to this channel, or None.""" + with self._connect() as conn: + row = conn.execute( + """SELECT r.id AS region_id, r.name, r.key_hex, r.is_default + FROM channel_scopes cs + JOIN regions r ON r.id = cs.region_id + WHERE cs.channel_idx = ?""", + (channel_idx,) + ).fetchone() + return dict(row) if row else None + + def get_all_channel_scopes(self) -> Dict[int, Dict]: + """Bulk-load the full channel->region mapping for UI rendering.""" + with self._connect() as conn: + rows = conn.execute( + """SELECT cs.channel_idx, r.id AS region_id, r.name, r.key_hex, r.is_default + FROM channel_scopes cs + JOIN regions r ON r.id = cs.region_id""" + ).fetchall() + return {r['channel_idx']: dict(r) for r in rows} + # ================================================================ # Channel Messages # ================================================================ diff --git a/app/meshcore/regions.py b/app/meshcore/regions.py new file mode 100644 index 0000000..c9206f0 --- /dev/null +++ b/app/meshcore/regions.py @@ -0,0 +1,53 @@ +""" +MeshCore flood-scope (region) helpers. + +Key derivation and name validation for the per-channel region-scope feature. +Kept free of Flask/DB imports so it can be unit-tested in isolation. + +Firmware references: +- Key: SHA256('#' + name)[:16] (TransportKeyStore::getAutoKeyFor) +- Name rule: '-', '$', '#', digits, or any byte >= 'A' (RegionMap::is_name_char) +- Name length: fits in a 31-char field (30 chars + NUL terminator) +""" + +import hashlib +from typing import Tuple + +MAX_NAME_LEN = 30 # firmware NodePrefs.default_scope_name[31] = 30 chars + NUL + +_ALLOWED_SINGLE_BYTES = (0x2d, 0x24, 0x23) # '-', '$', '#' + + +def is_valid_region_name(name: str) -> Tuple[bool, str]: + """Validate a region name against the firmware's RegionMap::is_name_char rule. + + Returns (ok, error_message). On success error_message is ''. + """ + if not isinstance(name, str) or not name: + return False, 'Name must be a non-empty string' + try: + encoded = name.encode('utf-8') + except UnicodeEncodeError: + return False, 'Name must be UTF-8 encodable' + if len(encoded) > MAX_NAME_LEN: + return False, f'Name too long (max {MAX_NAME_LEN} bytes)' + for b in encoded: + if b in _ALLOWED_SINGLE_BYTES: + continue + if 0x30 <= b <= 0x39: # digits + continue + if b >= 0x41: # any byte >= 'A' + continue + return False, f'Invalid character (byte 0x{b:02x})' + return True, '' + + +def derive_scope_key(name: str) -> bytes: + """Derive the 16-byte scope key: SHA256('#' + name)[:16].""" + payload = name if name.startswith('#') else '#' + name + return hashlib.sha256(payload.encode('utf-8')).digest()[:16] + + +def derive_scope_key_hex(name: str) -> str: + """Hex-encoded variant of derive_scope_key().""" + return derive_scope_key(name).hex() diff --git a/app/schema.sql b/app/schema.sql index d4203b1..769f7c8 100644 --- a/app/schema.sql +++ b/app/schema.sql @@ -43,6 +43,23 @@ CREATE TABLE IF NOT EXISTS channels ( updated_at TEXT NOT NULL DEFAULT (datetime('now')) ); +-- Region registry (user-curated MeshCore flood scopes) +CREATE TABLE IF NOT EXISTS regions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL UNIQUE, -- firmware-safe name, e.g. 'pl-ma' + key_hex TEXT NOT NULL, -- 32 hex chars = 16-byte scope key + is_default INTEGER NOT NULL DEFAULT 0, -- mirrors firmware CMD_GET_DEFAULT_FLOOD_SCOPE + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +-- Per-channel region mapping (absent row = no override; firmware default applies) +CREATE TABLE IF NOT EXISTS channel_scopes ( + channel_idx INTEGER PRIMARY KEY, + region_id INTEGER NOT NULL REFERENCES regions(id) ON DELETE CASCADE, + updated_at TEXT NOT NULL DEFAULT (datetime('now')) +); + -- Channel messages (replaces CHAN/SENT_CHAN from .msgs) CREATE TABLE IF NOT EXISTS channel_messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -188,6 +205,7 @@ CREATE INDEX IF NOT EXISTS idx_echoes_pkt ON echoes(pkt_payload); CREATE INDEX IF NOT EXISTS idx_adv_pubkey ON advertisements(public_key, timestamp); CREATE INDEX IF NOT EXISTS idx_contacts_name ON contacts(name); CREATE INDEX IF NOT EXISTS idx_cp_contact ON contact_paths(contact_pubkey, sort_order); +CREATE INDEX IF NOT EXISTS idx_regions_default ON regions(is_default) WHERE is_default = 1; -- ============================================================ -- Full-Text Search (FTS5) diff --git a/tests/test_regions.py b/tests/test_regions.py new file mode 100644 index 0000000..f6e8539 --- /dev/null +++ b/tests/test_regions.py @@ -0,0 +1,246 @@ +""" +Unit + integration tests for the per-channel region-scope data layer. + +Run: python -m pytest tests/test_regions.py -v +""" + +import sqlite3 +import tempfile +from pathlib import Path + +import pytest + +from app.database import Database +from app.meshcore.regions import ( + MAX_NAME_LEN, + derive_scope_key, + derive_scope_key_hex, + is_valid_region_name, +) + + +@pytest.fixture +def db(): + with tempfile.TemporaryDirectory() as tmp: + yield Database(Path(tmp) / 'test.db') + + +# ================================================================ +# Key derivation (known vectors) +# ================================================================ + +class TestKeyDerivation: + # Firmware rule: key = SHA256('#' + name)[:16] + # Vectors computed offline and baked in to catch regressions. + def test_pl(self): + assert derive_scope_key_hex('pl') == '89e07394d9523e8996cae464c7770516' + + def test_pl_ma(self): + assert derive_scope_key_hex('pl-ma') == '71a012b2fcfee9b6a29a28729236f1b8' + + def test_krakow(self): + assert derive_scope_key_hex('krakow') == '1482a54016edec3b8d13a879b7af62a3' + + def test_returns_16_bytes(self): + assert len(derive_scope_key('pl')) == 16 + + def test_hash_input_skips_existing_hash_prefix(self): + # '#pl' must produce the same key as 'pl' — firmware does not double-prefix. + assert derive_scope_key_hex('#pl') == derive_scope_key_hex('pl') + + +# ================================================================ +# Name validation (firmware RegionMap::is_name_char rule) +# ================================================================ + +class TestNameValidation: + @pytest.mark.parametrize('name', [ + 'pl', 'pl-ma', 'pl#test', '$EU', '999', 'Malopolska', 'a', + '-leading-dash-ok', 'UPPER', 'mixedCase', + # Firmware rule `c >= 'A'` (0x41) admits underscore (0x5F) too. + 'my_region', + ]) + def test_valid(self, name): + ok, err = is_valid_region_name(name) + assert ok, f'expected valid, got error: {err}' + + @pytest.mark.parametrize('name', [ + '', # empty + ' pl', # space (0x20) + 'my region', # embedded space + 'a.b', # dot (0x2E) + 'a,b', # comma (0x2C) + 'a/b', # slash (0x2F) + 'a:b', # colon (0x3A) + 'a+b', # plus (0x2B) + 'a@b', # at-sign (0x40) + 'a(b', # (0x28) + 'a*b', # (0x2A) + ]) + def test_invalid(self, name): + ok, _ = is_valid_region_name(name) + assert not ok, f'expected invalid for: {name!r}' + + def test_too_long_rejected(self): + too_long = 'a' * (MAX_NAME_LEN + 1) + ok, _ = is_valid_region_name(too_long) + assert not ok + + def test_at_length_limit_accepted(self): + at_limit = 'a' * MAX_NAME_LEN + ok, _ = is_valid_region_name(at_limit) + assert ok + + def test_non_string_rejected(self): + for bad in [None, 42, b'pl', ['pl']]: + ok, _ = is_valid_region_name(bad) + assert not ok + + def test_accented_chars_accepted(self): + # Firmware rule admits any byte >= 'A' (0x41), which includes all UTF-8 + # continuation bytes (>=0x80), so accented chars pass. + ok, _ = is_valid_region_name('Malopolska') + assert ok + ok, _ = is_valid_region_name('Kraków') + assert ok + + +# ================================================================ +# DB: region CRUD +# ================================================================ + +class TestRegionCrud: + def test_create_and_list(self, db): + rid = db.create_region('pl', derive_scope_key_hex('pl')) + assert rid > 0 + regions = db.list_regions() + assert len(regions) == 1 + assert regions[0]['name'] == 'pl' + assert regions[0]['key_hex'] == derive_scope_key_hex('pl') + assert regions[0]['is_default'] == 0 + + def test_duplicate_name_raises(self, db): + db.create_region('pl', derive_scope_key_hex('pl')) + with pytest.raises(sqlite3.IntegrityError): + db.create_region('pl', derive_scope_key_hex('pl')) + + def test_get_by_id_and_name(self, db): + rid = db.create_region('pl-ma', derive_scope_key_hex('pl-ma')) + by_id = db.get_region(rid) + by_name = db.get_region_by_name('pl-ma') + assert by_id and by_name + assert by_id['id'] == by_name['id'] == rid + + def test_get_missing_returns_none(self, db): + assert db.get_region(999) is None + assert db.get_region_by_name('missing') is None + + def test_delete(self, db): + rid = db.create_region('pl', derive_scope_key_hex('pl')) + assert db.delete_region(rid) is True + assert db.get_region(rid) is None + assert db.delete_region(rid) is False # already gone + + def test_list_ordered_by_name(self, db): + db.create_region('pl-ma', derive_scope_key_hex('pl-ma')) + db.create_region('pl', derive_scope_key_hex('pl')) + db.create_region('krakow', derive_scope_key_hex('krakow')) + names = [r['name'] for r in db.list_regions()] + assert names == ['krakow', 'pl', 'pl-ma'] + + +# ================================================================ +# DB: default region +# ================================================================ + +class TestDefaultRegion: + def test_no_default_initially(self, db): + assert db.get_default_region() is None + + def test_set_and_get_default(self, db): + rid = db.create_region('pl', derive_scope_key_hex('pl')) + db.set_default_region(rid) + d = db.get_default_region() + assert d is not None + assert d['id'] == rid + assert d['is_default'] == 1 + + def test_set_default_clears_previous(self, db): + a = db.create_region('pl', derive_scope_key_hex('pl')) + b = db.create_region('pl-ma', derive_scope_key_hex('pl-ma')) + db.set_default_region(a) + db.set_default_region(b) + # only one default + defaults = [r for r in db.list_regions() if r['is_default']] + assert len(defaults) == 1 + assert defaults[0]['id'] == b + + def test_set_default_none_clears_all(self, db): + rid = db.create_region('pl', derive_scope_key_hex('pl')) + db.set_default_region(rid) + db.set_default_region(None) + assert db.get_default_region() is None + + +# ================================================================ +# DB: channel_scopes mapping +# ================================================================ + +class TestChannelScopes: + def test_set_and_get(self, db): + rid = db.create_region('pl', derive_scope_key_hex('pl')) + db.set_channel_scope(3, rid) + scope = db.get_channel_scope(3) + assert scope is not None + assert scope['region_id'] == rid + assert scope['name'] == 'pl' + assert scope['key_hex'] == derive_scope_key_hex('pl') + + def test_get_missing_returns_none(self, db): + assert db.get_channel_scope(5) is None + + def test_set_none_clears(self, db): + rid = db.create_region('pl', derive_scope_key_hex('pl')) + db.set_channel_scope(3, rid) + db.set_channel_scope(3, None) + assert db.get_channel_scope(3) is None + + def test_upsert_replaces(self, db): + a = db.create_region('pl', derive_scope_key_hex('pl')) + b = db.create_region('pl-ma', derive_scope_key_hex('pl-ma')) + db.set_channel_scope(3, a) + db.set_channel_scope(3, b) + scope = db.get_channel_scope(3) + assert scope['region_id'] == b + + def test_cascade_on_region_delete(self, db): + rid = db.create_region('pl', derive_scope_key_hex('pl')) + db.set_channel_scope(3, rid) + db.set_channel_scope(4, rid) + db.delete_region(rid) + assert db.get_channel_scope(3) is None + assert db.get_channel_scope(4) is None + + def test_get_all_channel_scopes(self, db): + a = db.create_region('pl', derive_scope_key_hex('pl')) + b = db.create_region('pl-ma', derive_scope_key_hex('pl-ma')) + db.set_channel_scope(0, a) + db.set_channel_scope(3, b) + all_scopes = db.get_all_channel_scopes() + assert set(all_scopes.keys()) == {0, 3} + assert all_scopes[0]['name'] == 'pl' + assert all_scopes[3]['name'] == 'pl-ma' + + +# ================================================================ +# Schema presence +# ================================================================ + +class TestSchema: + def test_regions_and_channel_scopes_tables_exist(self, db): + with db._connect() as conn: + tables = {r[0] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall()} + assert 'regions' in tables + assert 'channel_scopes' in tables