mirror of
https://github.com/MarekWo/mc-webui.git
synced 2026-08-06 08:43:17 +02:00
feat(regions): add data layer for per-channel region scopes
Introduces the SQLite-backed region registry and channel->region mapping
that will drive the per-channel flood-scope feature. No UI or device
wiring yet; those land in subsequent PRs.
- schema.sql: new `regions` and `channel_scopes` tables + partial index
on the default flag.
- database.py: CRUD helpers for regions (create/list/get/delete/default)
and channel_scopes (set/get/bulk-load) with ON DELETE CASCADE.
- app/meshcore/regions.py: pure helpers for SHA256('#'+name)[:16] key
derivation and firmware-compatible name validation (mirrors the
`RegionMap::is_name_char` rule `c in {-,$,#} or c>='0' or c>='A'`).
- tests/test_regions.py: known SHA256 vectors, validator coverage
(incl. the firmware quirk that `_` and other 0x5B-0x60 chars are
admitted), and CRUD + cascade integration tests.
This commit is contained in:
MarekWo
@@ -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
|
||||
# ================================================================
|
||||
|
||||
@@ -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()
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user