adam/pyMC_Repeater
1
0
Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-08-08 09:52:55 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: add region/default-scope / cli commands update

Browse Source 
standardize default region on mesh.default_region (remove legacy region_default_scope usage)
add/align CLI support for owner.info, path.hash.mode, and loop.detect with validation
wire UI terminal get/set + autocomplete/help for owner.info, path.hash.mode, loop.detect
extend update_radio_config to persist owner_info for UI set owner.info
add and use shared packet utility for advert creation + default-region transport scoping
refactor repeater and room-server advert paths to reuse shared packet logic
This commit is contained in:
Rightup
2026-07-08 17:25:35 +01:00
parent 92310235ae
commit 1906f576bb
19 changed files with 1296 additions and 1021 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.yaml.example
+8
View File
@@ -311,6 +311,10 @@ mesh:
# true = allow unscoped flooding, false = deny flooding globally
unscoped_flood_allow: true
# Legacy alias for unscoped flood policy. Keep in sync with unscoped_flood_allow.
# true = allow unscoped flooding, false = deny
global_flood_allow: true
# Path hash mode for flood packets (0-hop): per-hop hash size in path encoding
# 0 = 1-byte hashes (legacy), 1 = 2-byte, 2 = 3-byte. Must match mesh convention.
# Affects originated adverts and any other flood packets sent by the repeater.
@@ -320,6 +324,10 @@ mesh:
# off = disabled, minimal = allow up to 3 self-hashes, moderate = allow up to 1, strict = allow 0
loop_detect: minimal
# Default flood scope for locally-originated flood adverts.
# Use null or "<null>" via CLI command "region default <null>" to clear.
default_region: null
# Multiple Identity Configuration (Optional)
# Define additional identities for the repeater to manage
# Each identity operates independently with its own key pair and configuration
repeater/data_acquisition/sqlite_handler.py
+21 -1
View File
@@ -2143,11 +2143,29 @@ class SQLiteHandler:
try:
with self._connect() as conn:
cursor = conn.execute("DELETE FROM adverts WHERE id = ?", (advert_id,))
self._neighbors_cache = {"timestamp": 0.0, "value": None}
return cursor.rowcount > 0
except Exception as e:
logger.error(f"Failed to delete advert: {e}")
return False
def delete_neighbors_by_pubkey_prefix(self, pubkey_prefix: Optional[str]) -> int:
"""Delete neighbor adverts by pubkey prefix (or all when prefix is None)."""
try:
with self._connect() as conn:
if pubkey_prefix is None:
cursor = conn.execute("DELETE FROM adverts")
else:
cursor = conn.execute(
"DELETE FROM adverts WHERE lower(pubkey) LIKE ?",
(f"{pubkey_prefix.lower()}%",),
)
self._neighbors_cache = {"timestamp": 0.0, "value": None}
return int(cursor.rowcount)
except Exception as e:
logger.error(f"Failed to delete neighbors by prefix: {e}")
raise
# ------------------------------------------------------------------
# Room Server Methods
# ------------------------------------------------------------------
@@ -2755,7 +2773,9 @@ class SQLiteHandler:
logger.error(f"Failed to count companion messages: {e}")
return 0
def companion_load_messages(self, companion_hash: str, limit: int = 100) -> Optional[List[Dict]]:
def companion_load_messages(
self, companion_hash: str, limit: int = 100
) -> Optional[List[Dict]]:
"""Load queued messages for a companion (oldest first for queue order).
Returns [] when the companion has no persisted messages, or None when
repeater/data_acquisition/storage_collector.py
+3
View File
@@ -551,6 +551,9 @@ class StorageCollector:
def delete_advert(self, advert_id: int) -> bool:
return self.sqlite_handler.delete_advert(advert_id)
def delete_neighbors_by_pubkey_prefix(self, pubkey_prefix: str | None) -> int:
return self.sqlite_handler.delete_neighbors_by_pubkey_prefix(pubkey_prefix)
def get_hardware_stats(self) -> Optional[dict]:
"""Get current hardware statistics"""
try:
repeater/data_acquisition/websocket_handler.py
+5 -1
View File
@@ -57,7 +57,11 @@ class PacketWebSocket(WebSocket):
try:
payload = jwt_handler.verify_jwt(token)
if payload:
if client_id and payload.get("client_id") and payload.get("client_id") != client_id:
if (
client_id
and payload.get("client_id")
and payload.get("client_id") != client_id
):
logger.warning("WebSocket connection rejected: client_id mismatch")
self.close(code=1008, reason="unauthorized")
return
repeater/handler_helpers/mesh_cli.py
+543 -21
View File
@@ -37,6 +37,112 @@ class MeshCLI:
# Get repeater config shortcut
self.repeater_config = config.get("repeater", {})
self.mesh_config = config.setdefault("mesh", {})
def _get_node_name(self) -> str:
"""Return the configured node name, preferring the newer key when present."""
return self.repeater_config.get("node_name") or self.repeater_config.get("name", "Unknown")
def _set_node_name(self, value: str) -> None:
"""Persist node name to both legacy and current config keys for compatibility."""
self.repeater_config["node_name"] = value
self.repeater_config["name"] = value
def _get_local_pubkey_hex(self) -> Optional[str]:
"""Return local node public key (hex) when available."""
try:
if self.identity and hasattr(self.identity, "get_public_key"):
pubkey = self.identity.get_public_key()
if isinstance(pubkey, (bytes, bytearray)):
return bytes(pubkey).hex().lower()
if isinstance(pubkey, str):
normalized = pubkey.strip().lower()
if normalized.startswith("0x"):
normalized = normalized[2:]
if normalized:
return normalized
except Exception as exc:
logger.debug("Unable to read local identity pubkey: %s", exc)
key = self.repeater_config.get("identity_key")
if isinstance(key, (bytes, bytearray)):
return bytes(key).hex().lower()
if isinstance(key, str):
normalized = key.strip().lower()
if normalized.startswith("0x"):
normalized = normalized[2:]
if normalized and all(ch in "0123456789abcdef" for ch in normalized):
return normalized
return None
def _is_local_pubkey(self, pubkey_hex: str) -> bool:
"""Return True when a discovery result pubkey matches the local node."""
candidate = (pubkey_hex or "").strip().lower()
if not candidate:
return False
if candidate.startswith("0x"):
candidate = candidate[2:]
local_pubkey = self._get_local_pubkey_hex()
if not local_pubkey:
return False
# Prefix-only discovery may return fewer bytes than full identity pubkey.
return local_pubkey.startswith(candidate) or candidate.startswith(local_pubkey)
def _auto_add_discovery_result(self, result: Dict[str, Any]) -> Dict[str, Any]:
"""Persist discovered neighbors automatically, excluding this node itself."""
enriched = dict(result)
pub_key = str(enriched.get("pub_key") or "").strip().lower()
if not pub_key:
return enriched
if self._is_local_pubkey(pub_key):
enriched["is_self"] = True
enriched["known_neighbor"] = True
return enriched
if not self.storage_handler:
return enriched
record_advert = getattr(self.storage_handler, "record_advert", None)
if not callable(record_advert):
return enriched
try:
import time
node_type = int(enriched.get("node_type", 0) or 0)
contact_type = {
1: "Chat Node",
2: "Repeater",
3: "Room Server",
}.get(node_type, "Unknown")
rssi = enriched.get("rssi")
snr = enriched.get("response_snr", enriched.get("snr"))
advert_record = {
"timestamp": time.time(),
"pubkey": pub_key,
"node_name": enriched.get("node_name"),
"is_repeater": node_type == 2,
"route_type": 2,
"contact_type": contact_type,
"latitude": None,
"longitude": None,
"rssi": int(rssi) if rssi is not None else None,
"snr": float(snr) if snr is not None else None,
"is_new_neighbor": True,
"zero_hop": True,
}
record_advert(advert_record)
enriched["known_neighbor"] = True
enriched["auto_added"] = True
except Exception as exc:
logger.debug("Auto-add discovery result failed for %s: %s", pub_key, exc)
return enriched
def handle_command(self, sender_pubkey: bytes, command: str, is_admin: bool) -> str:
@@ -119,6 +225,8 @@ class MeshCLI:
return self._cmd_neighbors()
elif command.startswith("neighbor.remove "):
return self._cmd_neighbor_remove(command)
elif command.startswith("discover.neighbors"):
return self._cmd_discover_neighbors(command)
# Temporary radio params
elif command.startswith("tempradio "):
@@ -152,7 +260,7 @@ class MeshCLI:
return self._help_detail(parts[1])
lines = [
"=== pyMC CLI Commands ===",
"=== openHop CLI Commands ===",
"",
"System:",
" reboot Restart the repeater service",
@@ -174,11 +282,14 @@ class MeshCLI:
" get repeat Repeat mode (on/off)",
" get lat / get lon GPS coordinates",
" get role Identity role",
" get owner.info Owner info text",
" get guest.password Guest password",
" get allow.read.only Read-only access setting",
" get advert.interval Advert interval (minutes)",
" get flood.advert.interval Flood advert interval (hours)",
" get flood.max Max flood hops",
" get path.hash.mode Flood advert path hash mode (0-2)",
" get loop.detect Flood loop detection mode",
" get rxdelay RX delay base",
" get txdelay TX delay factor",
" get direct.txdelay Direct TX delay factor",
@@ -192,6 +303,7 @@ class MeshCLI:
"Other:",
" neighbors List neighbors",
" neighbor.remove <key> Remove neighbor by pubkey",
" discover.neighbors Send zero-hop neighbor discovery",
" tempradio <freq> <bw> <sf> <cr> <timeout_mins>",
" setperm <pubkey> <perm> Set ACL permissions",
" log start|stop|erase Logging control",
@@ -216,10 +328,13 @@ class MeshCLI:
" set lat <deg> Latitude\n"
" set lon <deg> Longitude\n"
" set guest.password <pw> Guest password\n"
" set owner.info <text> Owner info text\n"
" set allow.read.only on|off Read-only access\n"
" set advert.interval <min> 60-240 minutes\n"
" set flood.advert.interval <hr> 3-168 hours\n"
" set flood.max <hops> Max flood hops (max 64)\n"
" set path.hash.mode <0-2> Path hash mode (0=1B,1=2B,2=3B)\n"
" set loop.detect <off|minimal|moderate|strict> Flood loop detection\n"
" set rxdelay <val> RX delay base (>=0)\n"
" set txdelay <val> TX delay factor (>=0)\n"
" set direct.txdelay <val> Direct TX delay (>=0)\n"
@@ -242,6 +357,7 @@ class MeshCLI:
" freq: 300-2500 MHz, bw: 7-500 kHz, sf: 5-12, cr: 5-8"
),
"neighbors": "List known neighbor nodes from the routing table.",
"discover.neighbors": "Send a neighbor discovery request.",
"setperm": "setperm <pubkey_hex> <permission_int> \u2014 Set ACL permissions for a node.",
"log": "log start|stop|erase \u2014 Control logging.",
}
@@ -372,8 +488,7 @@ class MeshCLI:
return f"> {af}"
elif param == "name":
name = self.repeater_config.get("name", "Unknown")
return f"> {name}"
return f"> {self._get_node_name()}"
elif param == "repeat":
mode = self.repeater_config.get("mode", "forward")
@@ -426,6 +541,10 @@ class MeshCLI:
guest_pw = self.config.get("security", {}).get("guest_password", "")
return f"> {guest_pw}"
elif param == "owner.info":
owner_info = self.repeater_config.get("owner_info", "")
return f"> {owner_info}"
elif param == "allow.read.only":
allow = self.config.get("security", {}).get("allow_read_only", False)
return f"> {'on' if allow else 'off'}"
@@ -442,6 +561,14 @@ class MeshCLI:
max_flood = self.repeater_config.get("max_flood_hops", 64)
return f"> {max_flood}"
elif param == "path.hash.mode":
path_hash_mode = self.mesh_config.get("path_hash_mode", 0)
return f"> {path_hash_mode}"
elif param == "loop.detect":
loop_detect = self.mesh_config.get("loop_detect", "off")
return f"> {loop_detect}"
elif param == "rxdelay":
delay = self.repeater_config.get("rx_delay_base", 0.0)
return f"> {delay}"
@@ -487,7 +614,7 @@ class MeshCLI:
return "OK"
elif key == "name":
self.repeater_config["node_name"] = value
self._set_node_name(value)
saved, _ = self.config_manager.save_to_file()
self.config_manager.live_update_daemon(["repeater"])
return "OK"
@@ -551,6 +678,12 @@ class MeshCLI:
self.config_manager.live_update_daemon(["security"])
return "OK"
elif key == "owner.info":
self.repeater_config["owner_info"] = value.replace("|", "\n")
saved, _ = self.config_manager.save_to_file()
self.config_manager.live_update_daemon(["repeater"])
return "OK"
elif key == "allow.read.only":
if "security" not in self.config:
self.config["security"] = {}
@@ -586,6 +719,24 @@ class MeshCLI:
self.config_manager.live_update_daemon(["repeater"])
return "OK"
elif key == "path.hash.mode":
mode = int(value)
if mode not in (0, 1, 2):
return "Error: path.hash.mode must be 0, 1, or 2"
self.mesh_config["path_hash_mode"] = mode
saved, _ = self.config_manager.save_to_file()
self.config_manager.live_update_daemon(["mesh"])
return "OK"
elif key == "loop.detect":
mode = str(value).strip().lower()
if mode not in ("off", "minimal", "moderate", "strict"):
return "Error: loop.detect must be off, minimal, moderate, or strict"
self.mesh_config["loop_detect"] = mode
saved, _ = self.config_manager.save_to_file()
self.config_manager.live_update_daemon(["mesh"])
return "OK"
elif key == "rxdelay":
delay = float(value)
if delay < 0:
@@ -665,22 +816,336 @@ class MeshCLI:
# ==================== Region Commands ====================
def _cmd_region(self, command: str) -> str:
"""Handle region commands."""
"""Handle region commands with MeshCore-compatible response shapes."""
parts = command.split()
if len(parts) == 1:
return "Error: Region commands not implemented in Python repeater"
return self._region_export_tree()
subcommand = parts[1]
if subcommand == "load":
return "Error: Region commands not implemented"
elif subcommand == "save":
return "Error: Region commands not implemented"
elif subcommand in ("allowf", "denyf", "get", "home", "put", "remove"):
return "Error: Region commands not implemented"
else:
return "Err - ??"
return "Err - region load not supported"
if subcommand == "def":
return "Err - region def not supported"
if subcommand == "save":
return self._region_save()
if subcommand == "allowf" and len(parts) >= 3:
return self._region_set_flood(parts[2], allow=True)
if subcommand == "denyf" and len(parts) >= 3:
return self._region_set_flood(parts[2], allow=False)
if subcommand == "get" and len(parts) >= 3:
return self._region_get(parts[2])
if subcommand == "home":
if len(parts) >= 3:
return self._region_home_set(parts[2])
return self._region_home_get()
if subcommand == "default":
if len(parts) >= 3:
return self._region_default_set(parts[2])
return self._region_default_get()
if subcommand == "put" and len(parts) >= 3:
parent = parts[3] if len(parts) >= 4 else "*"
return self._region_put(parts[2], parent)
if subcommand == "remove" and len(parts) >= 3:
return self._region_remove(parts[2])
if subcommand == "list" and len(parts) >= 3:
return self._region_list(parts[2])
return "Err - ??"
def _region_storage_available(self) -> bool:
return bool(
self.storage_handler
and hasattr(self.storage_handler, "get_transport_keys")
and callable(getattr(self.storage_handler, "get_transport_keys"))
)
def _region_load_records(self) -> list[dict]:
if not self._region_storage_available():
return []
records = self.storage_handler.get_transport_keys()
return records if isinstance(records, list) else []
@staticmethod
def _region_display_name(raw_name: str) -> str:
name = str(raw_name or "").strip()
if name.startswith("#"):
return name[1:]
return name
def _region_find_prefix(self, query: str) -> Optional[dict]:
q = str(query or "").strip()
if not q:
return None
if q == "*":
return {
"id": 0,
"name": "*",
"display_name": "*",
"flood_policy": "allow" if self._region_unscoped_allow() else "deny",
"parent_id": None,
}
q_lower = q.lower()
for rec in self._region_load_records():
display = self._region_display_name(rec.get("name", ""))
if display.lower().startswith(q_lower):
return {**rec, "display_name": display}
return None
def _region_find_exact(self, query: str) -> Optional[dict]:
q = str(query or "").strip()
if not q:
return None
if q == "*":
return {
"id": 0,
"name": "*",
"display_name": "*",
"flood_policy": "allow" if self._region_unscoped_allow() else "deny",
"parent_id": None,
}
q_lower = q.lower()
for rec in self._region_load_records():
display = self._region_display_name(rec.get("name", ""))
if display.lower() == q_lower:
return {**rec, "display_name": display}
return None
def _region_unscoped_allow(self) -> bool:
return bool(
self.mesh_config.get(
"unscoped_flood_allow",
self.mesh_config.get("global_flood_allow", True),
)
)
def _region_set_unscoped_allow(self, allow: bool) -> bool:
self.mesh_config["unscoped_flood_allow"] = bool(allow)
self.mesh_config["global_flood_allow"] = bool(allow)
save_result = self.config_manager.save_to_file()
saved = save_result[0] if isinstance(save_result, tuple) else bool(save_result)
self.config_manager.live_update_daemon(["mesh"])
return bool(saved)
def _region_get_default_name(self) -> Optional[str]:
default_name = self.mesh_config.get("default_region")
text = str(default_name).strip() if default_name is not None else ""
return text or None
def _region_set_default_name(self, value: Optional[str]) -> bool:
self.mesh_config["default_region"] = value
save_result = self.config_manager.save_to_file()
saved = save_result[0] if isinstance(save_result, tuple) else bool(save_result)
self.config_manager.live_update_daemon(["mesh"])
return bool(saved)
def _region_export_tree(self) -> str:
records = self._region_load_records()
children_by_parent: Dict[int, list[dict]] = {}
for rec in records:
parent_id = rec.get("parent_id")
parent_key = int(parent_id) if isinstance(parent_id, int) and parent_id > 0 else 0
children_by_parent.setdefault(parent_key, []).append(rec)
for parent_list in children_by_parent.values():
parent_list.sort(key=lambda r: str(r.get("name", "")).lower())
home_name = str(self.repeater_config.get("region_home") or "").strip().lower()
lines: list[str] = []
def append_line(indent: int, display_name: str, flood_policy: str):
home_mark = "^" if home_name and display_name.lower() == home_name else ""
flood_mark = " F" if flood_policy == "allow" else ""
lines.append(f"{' ' * indent}{display_name}{home_mark}{flood_mark}")
append_line(0, "*", "allow" if self._region_unscoped_allow() else "deny")
def walk(parent_id: int, indent: int):
for rec in children_by_parent.get(parent_id, []):
display_name = self._region_display_name(rec.get("name", ""))
append_line(indent, display_name, str(rec.get("flood_policy", "deny")))
walk(int(rec.get("id", 0)), indent + 1)
walk(0, 1)
return "\n".join(lines)
def _region_save(self) -> str:
save_result = self.config_manager.save_to_file()
saved = save_result[0] if isinstance(save_result, tuple) else bool(save_result)
return "OK" if saved else "Err - save failed"
def _region_set_flood(self, name_prefix: str, allow: bool) -> str:
region = self._region_find_prefix(name_prefix)
if not region:
return "Err - unknown region"
if region.get("id") == 0:
return "OK" if self._region_set_unscoped_allow(allow) else "Err - save failed"
update_fn = getattr(self.storage_handler, "update_transport_key", None)
if not callable(update_fn):
return "Error: Region commands not supported by storage backend"
ok = update_fn(int(region["id"]), flood_policy="allow" if allow else "deny")
return "OK" if ok else "Err - unknown region"
def _region_get(self, name_prefix: str) -> str:
region = self._region_find_prefix(name_prefix)
if not region:
return "Err - unknown region"
display_name = str(
region.get("display_name") or self._region_display_name(region.get("name", ""))
)
flood_suffix = "F" if region.get("flood_policy") == "allow" else ""
parent_name = None
parent_id = region.get("parent_id")
if isinstance(parent_id, int) and parent_id > 0:
for rec in self._region_load_records():
if int(rec.get("id", -1)) == parent_id:
parent_name = self._region_display_name(rec.get("name", ""))
break
if parent_name:
return f" {display_name} ({parent_name}) {flood_suffix}".rstrip()
return f" {display_name} {flood_suffix}".rstrip()
def _region_home_get(self) -> str:
home = str(self.repeater_config.get("region_home") or "").strip()
return f" home is {home or '*'}"
def _region_home_set(self, name_prefix: str) -> str:
region = self._region_find_prefix(name_prefix)
if not region:
return "Err - unknown region"
display_name = str(region.get("display_name") or "*")
self.repeater_config["region_home"] = display_name
save_result = self.config_manager.save_to_file()
saved = save_result[0] if isinstance(save_result, tuple) else bool(save_result)
return f" home is now {display_name}" if saved else "Err - save failed"
def _region_default_get(self) -> str:
default_region = self._region_get_default_name()
if default_region is None:
return " default scope is <null>"
return f" default scope is {default_region}"
def _region_default_set(self, value: str) -> str:
text = str(value or "").strip()
if text == "<null>":
saved = self._region_set_default_name(None)
return " default scope is now <null>" if saved else "Err - save failed"
region = self._region_find_prefix(text)
if region:
display_name = str(region.get("display_name") or text)
if region.get("id") not in (None, 0):
update_fn = getattr(self.storage_handler, "update_transport_key", None)
if callable(update_fn):
update_fn(int(region["id"]), flood_policy="allow")
saved = self._region_set_default_name(display_name)
return f" default scope is now {display_name}" if saved else "Err - save failed"
put_result = self._region_put(text, "*")
if not put_result.startswith("OK"):
return "Err - region table full"
saved = self._region_set_default_name(text)
return f" default scope is now {text}" if saved else "Err - save failed"
def _region_put(self, name: str, parent_name: str) -> str:
region_name = str(name or "").strip()
if not region_name:
return "Err - unable to put"
parent = self._region_find_prefix(parent_name)
if not parent:
return "Err - unknown parent"
parent_id = int(parent.get("id", 0))
parent_storage_id = None if parent_id == 0 else parent_id
existing = self._region_find_exact(region_name)
if existing and existing.get("id") != 0:
update_fn = getattr(self.storage_handler, "update_transport_key", None)
if not callable(update_fn):
return "Err - unable to put"
ok = update_fn(
int(existing["id"]),
flood_policy="allow",
parent_id=parent_storage_id,
)
return "OK - (flood allowed)" if ok else "Err - unable to put"
create_fn = getattr(self.storage_handler, "create_transport_key", None)
if not callable(create_fn):
return "Err - unable to put"
key_id = create_fn(
region_name,
"allow",
None,
parent_storage_id,
None,
)
return "OK - (flood allowed)" if key_id else "Err - unable to put"
def _region_remove(self, name: str) -> str:
region = self._region_find_exact(name)
if not region or region.get("id") == 0:
return "Err - not found"
region_id = int(region["id"])
for rec in self._region_load_records():
if int(rec.get("parent_id") or 0) == region_id:
return "Err - not empty"
delete_fn = getattr(self.storage_handler, "delete_transport_key", None)
if not callable(delete_fn):
return "Err - not found"
ok = delete_fn(region_id)
if not ok:
return "Err - not found"
removed_name = str(region.get("display_name") or "")
if (
str(self.repeater_config.get("region_home") or "").strip().lower()
== removed_name.lower()
):
self.repeater_config["region_home"] = ""
default_name = self._region_get_default_name()
if str(default_name or "").strip().lower() == removed_name.lower():
self.mesh_config["default_region"] = None
return "OK"
def _region_list(self, filter_name: str) -> str:
mode = str(filter_name or "").strip().lower()
if mode not in ("allowed", "denied"):
return "Err - use 'allowed' or 'denied'"
names: list[str] = []
unscoped_allowed = self._region_unscoped_allow()
if (mode == "allowed" and unscoped_allowed) or (mode == "denied" and not unscoped_allowed):
names.append("*")
records = sorted(
self._region_load_records(),
key=lambda r: self._region_display_name(r.get("name", "")).lower(),
)
for rec in records:
flood_policy = str(rec.get("flood_policy", "deny")).lower()
allowed = flood_policy == "allow"
if (mode == "allowed" and allowed) or (mode == "denied" and not allowed):
names.append(self._region_display_name(rec.get("name", "")))
return ",".join(names) if names else "-none-"
# ==================== Neighbor Commands ====================
@@ -695,15 +1160,15 @@ class MeshCLI:
if not neighbors:
return "No neighbors discovered yet"
# Filter to only show repeaters and zero hop nodes
# Match MeshCore behavior: show only zero-hop repeaters.
filtered_neighbors = {
pubkey: info
for pubkey, info in neighbors.items()
if info.get("is_repeater", False) or info.get("zero_hop", False)
if info.get("is_repeater", False) and info.get("zero_hop", False)
}
if not filtered_neighbors:
return "No repeaters or zero hop neighbors discovered yet"
return "No zero hop repeaters discovered yet"
# Format output similar to C++ version
# Format: "<pubkey_prefix> heard Xs ago"
@@ -731,14 +1196,71 @@ class MeshCLI:
def _cmd_neighbor_remove(self, command: str) -> str:
"""Remove a neighbor."""
pubkey_hex = command[16:].strip()
raw_suffix = command[16:]
pubkey_hex = raw_suffix.strip()
if not pubkey_hex:
# Keep MeshCore parity: plain empty is invalid, whitespace-only means remove all.
if raw_suffix == "":
return "ERR: Missing pubkey"
# TODO: Remove neighbor from routing table
logger.info(f"neighbor.remove: {pubkey_hex}")
return "Error: Not yet implemented"
if not self.storage_handler:
return "Error: Storage not available"
delete_fn = getattr(self.storage_handler, "delete_neighbors_by_pubkey_prefix", None)
if not callable(delete_fn):
return "Error: neighbor.remove not supported by storage backend"
try:
if pubkey_hex == "":
delete_fn(None)
return "OK"
if any(ch not in "0123456789abcdefABCDEF" for ch in pubkey_hex):
return "ERR: bad pubkey"
delete_fn(pubkey_hex)
return "OK"
except Exception as e:
logger.error(f"neighbor.remove failed: {e}", exc_info=True)
return f"Error: {e}"
def _cmd_discover_neighbors(self, command: str) -> str:
"""Send a discovery request for nearby repeaters."""
sub = command[18:]
if sub.strip():
return "Err - discover.neighbors has no options"
daemon_instance = getattr(self.config_manager, "daemon", None)
discovery_helper = getattr(daemon_instance, "discovery_helper", None)
if not discovery_helper:
return "Error: Discovery helper not available"
import asyncio
loop = self._event_loop
if loop is None:
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = None
if loop is None or not loop.is_running():
return "Error: Event loop not available"
try:
discovery_helper.cleanup_sessions()
session = discovery_helper.create_session(
timeout=5,
filter_mask=(1 << 2),
since=0,
prefix_only=False,
result_enricher=self._auto_add_discovery_result,
)
loop.call_soon_threadsafe(discovery_helper.start_session_task, session["session_id"])
return "OK - Discover sent"
except Exception as e:
logger.error(f"discover.neighbors failed: {e}", exc_info=True)
return f"Error: {e}"
# ==================== Temporary Radio Commands ====================
repeater/handler_helpers/repeater_cli.py
-700
View File
@@ -1,700 +0,0 @@
"""
Mesh CLI Handler
Handles administrative commands sent to repeaters and room servers via TXT_MSG packets.
Only users with admin permissions (via ACL) can execute these commands.
"""
import logging
from pathlib import Path
from typing import Any, Callable, Dict
logger = logging.getLogger(__name__)
class MeshCLI:
"""
CLI command handler for mesh node administration (repeaters and room servers).
Commands follow the format: XX|command params
where XX is an optional sequence number that gets echoed in the reply.
"""
def __init__(
self,
config_path: str,
config: Dict[str, Any],
save_config_callback: Callable,
identity_type: str = "repeater",
enable_regions: bool = True,
):
"""
Initialize the CLI handler.
Args:
config_path: Path to the config.yaml file
config: Current configuration dictionary
save_config_callback: Callback to save config changes
identity_type: Type of identity ('repeater' or 'room_server')
enable_regions: Whether to enable region commands (only for repeaters)
"""
self.config_path = Path(config_path)
self.config = config
self.save_config = save_config_callback
self.identity_type = identity_type
self.enable_regions = enable_regions
# Get repeater config shortcut
self.repeater_config = config.get("repeater", {})
def handle_command(self, sender_pubkey: bytes, command: str, is_admin: bool) -> str:
"""
Handle an incoming command from a client.
Args:
sender_pubkey: Public key of sender
command: Command string (may include XX| prefix)
is_admin: Whether sender has admin permissions
Returns:
Reply string to send back to sender
"""
# Check admin permission first
if not is_admin:
return "Error: Admin permission required"
logger.debug(f"handle_command received: '{command}' (len={len(command)})")
# Extract optional sequence prefix (XX|)
prefix = ""
if len(command) > 4 and command[2] == "|":
prefix = command[:3]
command = command[3:]
logger.debug(f"Extracted prefix: '{prefix}', remaining command: '{command}'")
# Strip leading/trailing whitespace
command = command.strip()
logger.debug(f"After strip: '{command}'")
# Route to appropriate handler
reply = self._route_command(command)
# Add prefix back to reply if present
if prefix:
return prefix + reply
return reply
def _route_command(self, command: str) -> str:
"""Route command to appropriate handler method."""
# Help
if command == "help" or command.startswith("help "):
return self._cmd_help(command)
# System commands
elif command == "reboot":
return self._cmd_reboot()
elif command == "advert":
return self._cmd_advert()
elif command.startswith("clock"):
return self._cmd_clock(command)
elif command.startswith("time "):
return self._cmd_time(command)
elif command == "start ota":
return "Error: OTA not supported in Python repeater"
elif command.startswith("password "):
return self._cmd_password(command)
elif command == "clear stats":
return self._cmd_clear_stats()
elif command == "ver":
return self._cmd_version()
# Get commands
elif command.startswith("get "):
return self._cmd_get(command[4:])
# Set commands
elif command.startswith("set "):
return self._cmd_set(command[4:])
# ACL commands
elif command.startswith("setperm "):
return self._cmd_setperm(command)
elif command == "get acl":
return "Error: Use 'get acl' via serial console only"
# Region commands (repeaters only)
elif command.startswith("region"):
if self.enable_regions:
return self._cmd_region(command)
else:
return "Error: Region commands not available for room servers"
# Neighbor commands
elif command == "neighbors":
return self._cmd_neighbors()
elif command.startswith("neighbor.remove "):
return self._cmd_neighbor_remove(command)
# Temporary radio params
elif command.startswith("tempradio "):
return self._cmd_tempradio(command)
# Sensor commands
elif command.startswith("sensor "):
return "Error: Sensor commands not implemented in Python repeater"
# GPS commands
elif command.startswith("gps"):
return "Error: GPS commands not implemented in Python repeater"
# Logging commands
elif command.startswith("log "):
return self._cmd_log(command)
# Statistics commands
elif command.startswith("stats-"):
return "Error: Stats commands not fully implemented yet"
else:
return "Unknown command"
# ==================== Help Command ====================
def _cmd_help(self, command: str) -> str:
"""Show available commands or detailed help for a specific command."""
parts = command.split(None, 1)
if len(parts) == 2:
return self._help_detail(parts[1])
lines = [
"=== pyMC CLI Commands ===",
"",
"System:",
" reboot Restart the repeater service",
" advert Send self advertisement",
" clock Show current UTC time",
" clock sync Sync clock (no-op, uses system time)",
" ver Show version info",
" password <pw> Change admin password",
" clear stats Clear statistics",
"",
"Get:",
" get name Node name",
" get radio Radio params (freq,bw,sf,cr)",
" get freq Frequency (MHz)",
" get tx TX power",
" get af Airtime factor",
" get repeat Repeat mode (on/off)",
" get lat / get lon GPS coordinates",
" get role Identity role",
" get guest.password Guest password",
" get allow.read.only Read-only access setting",
" get advert.interval Advert interval (minutes)",
" get flood.advert.interval Flood advert interval (hours)",
" get flood.max Max flood hops",
" get rxdelay RX delay base",
" get txdelay TX delay factor",
" get direct.txdelay Direct TX delay factor",
" get multi.acks Multi-ack count",
" get int.thresh Interference threshold",
" get agc.reset.interval AGC reset interval",
"",
"Set: (use 'help set' for details)",
" set <param> <value>",
"",
"Other:",
" neighbors List neighbors",
" neighbor.remove <key> Remove neighbor by pubkey",
" tempradio <freq> <bw> <sf> <cr> <timeout_mins>",
" setperm <pubkey> <perm> Set ACL permissions",
" log start|stop|erase Logging control",
]
if self.enable_regions:
lines.append(" region ... Region commands")
lines += ["", "Type 'help <command>' for details on a specific command."]
return "\n".join(lines)
def _help_detail(self, topic: str) -> str:
"""Return detailed help for a specific command topic."""
topic = topic.strip()
details = {
"set": (
"Set commands — set <param> <value>:\n"
" set name <name> Set node name\n"
" set radio <f> <bw> <sf> <cr> Set radio (restart required)\n"
" set freq <mhz> Set frequency (restart required)\n"
" set tx <power> Set TX power\n"
" set af <factor> Airtime factor\n"
" set repeat on|off Enable/disable repeating\n"
" set lat <deg> Latitude\n"
" set lon <deg> Longitude\n"
" set guest.password <pw> Guest password\n"
" set allow.read.only on|off Read-only access\n"
" set advert.interval <min> 60-240 minutes\n"
" set flood.advert.interval <hr> 3-168 hours\n"
" set flood.max <hops> Max flood hops (max 64)\n"
" set rxdelay <val> RX delay base (>=0)\n"
" set txdelay <val> TX delay factor (>=0)\n"
" set direct.txdelay <val> Direct TX delay (>=0)\n"
" set multi.acks <n> Multi-ack count\n"
" set int.thresh <dbm> Interference threshold\n"
" set agc.reset.interval <n> AGC reset (rounded to x4)"
),
"get": "Get commands — type 'help' to see all 'get' parameters.",
"reboot": "Restart the repeater service via systemd.",
"advert": "Trigger a self-advertisement flood packet.",
"clock": "'clock' shows UTC time. 'clock sync' is a no-op (system time used).",
"ver": "Show repeater version and identity type.",
"password": "password <new_password> — Change the admin password.",
"tempradio": (
"tempradio <freq_mhz> <bw_khz> <sf> <cr> <timeout_mins>\n"
" Apply temporary radio parameters that revert after timeout.\n"
" freq: 300-2500 MHz, bw: 7-500 kHz, sf: 5-12, cr: 5-8"
),
"neighbors": "List known neighbor nodes from the routing table.",
"setperm": "setperm <pubkey_hex> <permission_int> — Set ACL permissions for a node.",
"log": "log start|stop|erase — Control logging.",
}
return details.get(topic, f"No detailed help for '{topic}'. Type 'help' for command list.")
# ==================== System Commands ==
def _cmd_reboot(self) -> str:
"""Reboot the repeater process."""
from repeater.service_utils import restart_service
logger.warning("Reboot command received via repeater CLI")
success, message = restart_service()
if success:
return f"OK - {message}"
else:
return f"Error: {message}"
def _cmd_advert(self) -> str:
"""Send self advertisement."""
logger.info("Advert command received")
# TODO: Trigger advertisement through packet handler
return "Error: Not yet implemented"
def _cmd_clock(self, command: str) -> str:
"""Handle clock commands."""
if command == "clock":
# Display current time
import datetime
dt = datetime.datetime.now(datetime.timezone.utc)
return f"{dt.hour:02d}:{dt.minute:02d} - {dt.day}/{dt.month}/{dt.year} UTC"
elif command == "clock sync":
# Clock sync happens automatically via sender_timestamp in protocol
return "OK - clock sync not needed (system time used)"
else:
return "Unknown clock command"
def _cmd_time(self, command: str) -> str:
"""Set time - not supported in Python (use system time)."""
return "Error: Time setting not supported (system time is used)"
def _cmd_password(self, command: str) -> str:
"""Change admin password."""
new_password = command[9:].strip()
if not new_password:
return "Error: Password cannot be empty"
# Update security config
if "security" not in self.config:
self.config["security"] = {}
self.config["security"]["password"] = new_password
# Save config
try:
self.save_config()
return f"password now: {new_password}"
except Exception as e:
logger.error(f"Failed to save password: {e}")
return "Error: Failed to save password"
def _cmd_clear_stats(self) -> str:
"""Clear statistics."""
# TODO: Implement stats clearing
return "Error: Not yet implemented"
def _cmd_version(self) -> str:
"""Get version information."""
role = "room_server" if self.identity_type == "room_server" else "repeater"
version = self.config.get("version", "1.0.0")
return f"pyMC_{role} v{version}"
# ==================== Get Commands ====================
def _cmd_get(self, param: str) -> str:
"""Handle get commands."""
param = param.strip()
logger.debug(f"_cmd_get called with param: '{param}' (len={len(param)})")
if param == "af":
af = self.repeater_config.get("airtime_factor", 1.0)
return f"> {af}"
elif param == "name":
name = self.repeater_config.get("name", "Unknown")
return f"> {name}"
elif param == "repeat":
mode = self.repeater_config.get("mode", "forward")
return f"> {'on' if mode == 'forward' else 'off'}"
elif param == "lat":
lat = self.repeater_config.get("latitude", 0.0)
return f"> {lat}"
elif param == "lon":
lon = self.repeater_config.get("longitude", 0.0)
return f"> {lon}"
elif param == "radio":
radio = self.config.get("radio", {})
freq_hz = radio.get("frequency", 915000000)
bw_hz = radio.get("bandwidth", 125000)
sf = radio.get("spreading_factor", 7)
cr = radio.get("coding_rate", 5)
# Convert Hz to MHz for freq, Hz to kHz for bandwidth (match C++ ftoa output)
freq_mhz = freq_hz / 1_000_000.0
bw_khz = bw_hz / 1_000.0
return f"> {freq_mhz},{bw_khz},{sf},{cr}"
elif param == "freq":
freq_hz = self.config.get("radio", {}).get("frequency", 915000000)
freq_mhz = freq_hz / 1_000_000.0
return f"> {freq_mhz}"
elif param == "tx":
power = self.config.get("radio", {}).get("tx_power", 20)
return f"> {power}"
elif param == "public.key":
# TODO: Get from identity
return "Error: Not yet implemented"
elif param == "role":
role = "room_server" if self.identity_type == "room_server" else "repeater"
return f"> {role}"
elif param == "guest.password":
guest_pw = self.config.get("security", {}).get("guest_password", "")
return f"> {guest_pw}"
elif param == "allow.read.only":
allow = self.config.get("security", {}).get("allow_read_only", False)
return f"> {'on' if allow else 'off'}"
elif param == "advert.interval":
interval = self.repeater_config.get("advert_interval_minutes", 120)
return f"> {interval}"
elif param == "flood.advert.interval":
interval = self.repeater_config.get("flood_advert_interval_hours", 24)
return f"> {interval}"
elif param == "flood.max":
max_flood = self.repeater_config.get("max_flood_hops", 64)
return f"> {max_flood}"
elif param == "rxdelay":
delay = self.repeater_config.get("rx_delay_base", 0.0)
return f"> {delay}"
elif param == "txdelay":
delay = self.repeater_config.get("tx_delay_factor", 1.0)
return f"> {delay}"
elif param == "direct.txdelay":
delay = self.repeater_config.get("direct_tx_delay_factor", 0.5)
return f"> {delay}"
elif param == "multi.acks":
acks = self.repeater_config.get("multi_acks", 0)
return f"> {acks}"
elif param == "int.thresh":
thresh = self.repeater_config.get("interference_threshold", -120)
return f"> {thresh}"
elif param == "agc.reset.interval":
interval = self.repeater_config.get("agc_reset_interval", 0)
return f"> {interval}"
else:
return f"??: {param}"
# ==================== Set Commands ====================
def _cmd_set(self, param: str) -> str:
"""Handle set commands."""
parts = param.split(None, 1)
if len(parts) < 2:
return "Error: Missing value"
key, value = parts[0], parts[1]
try:
if key == "af":
self.repeater_config["airtime_factor"] = float(value)
self.save_config()
return "OK"
elif key == "name":
self.repeater_config["name"] = value
self.save_config()
return "OK"
elif key == "repeat":
self.repeater_config["mode"] = "forward" if value.lower() == "on" else "monitor"
self.save_config()
return f"OK - repeat is now {'ON' if self.repeater_config['mode'] == 'forward' else 'OFF'}"
elif key == "lat":
self.repeater_config["latitude"] = float(value)
self.save_config()
return "OK"
elif key == "lon":
self.repeater_config["longitude"] = float(value)
self.save_config()
return "OK"
elif key == "radio":
# Format: freq bw sf cr
radio_parts = value.split()
if len(radio_parts) != 4:
return "Error: Expected freq bw sf cr"
if "radio" not in self.config:
self.config["radio"] = {}
self.config["radio"]["frequency"] = float(radio_parts[0])
self.config["radio"]["bandwidth"] = float(radio_parts[1])
self.config["radio"]["spreading_factor"] = int(radio_parts[2])
self.config["radio"]["coding_rate"] = int(radio_parts[3])
self.save_config()
return "OK - restart repeater to apply"
elif key == "freq":
if "radio" not in self.config:
self.config["radio"] = {}
self.config["radio"]["frequency"] = float(value)
self.save_config()
return "OK - restart repeater to apply"
elif key == "tx":
if "radio" not in self.config:
self.config["radio"] = {}
self.config["radio"]["tx_power"] = int(value)
self.save_config()
return "OK"
elif key == "guest.password":
if "security" not in self.config:
self.config["security"] = {}
self.config["security"]["guest_password"] = value
self.save_config()
return "OK"
elif key == "allow.read.only":
if "security" not in self.config:
self.config["security"] = {}
self.config["security"]["allow_read_only"] = value.lower() == "on"
self.save_config()
return "OK"
elif key == "advert.interval":
mins = int(value)
if mins > 0 and (mins < 60 or mins > 240):
return "Error: interval range is 60-240 minutes"
self.repeater_config["advert_interval_minutes"] = mins
self.save_config()
return "OK"
elif key == "flood.advert.interval":
hours = int(value)
if (hours > 0 and hours < 3) or hours > 168:
return "Error: interval range is 3-168 hours"
self.repeater_config["flood_advert_interval_hours"] = hours
self.save_config()
return "OK"
elif key == "flood.max":
max_val = int(value)
if max_val > 64:
return "Error: max 64"
self.repeater_config["max_flood_hops"] = max_val
self.save_config()
return "OK"
elif key == "rxdelay":
delay = float(value)
if delay < 0:
return "Error: cannot be negative"
self.repeater_config["rx_delay_base"] = delay
self.save_config()
return "OK"
elif key == "txdelay":
delay = float(value)
if delay < 0:
return "Error: cannot be negative"
self.repeater_config["tx_delay_factor"] = delay
self.save_config()
return "OK"
elif key == "direct.txdelay":
delay = float(value)
if delay < 0:
return "Error: cannot be negative"
self.repeater_config["direct_tx_delay_factor"] = delay
self.save_config()
return "OK"
elif key == "multi.acks":
self.repeater_config["multi_acks"] = int(value)
self.save_config()
return "OK"
elif key == "int.thresh":
self.repeater_config["interference_threshold"] = int(value)
self.save_config()
return "OK"
elif key == "agc.reset.interval":
interval = int(value)
# Round to nearest multiple of 4
rounded = (interval // 4) * 4
self.repeater_config["agc_reset_interval"] = rounded
self.save_config()
return f"OK - interval rounded to {rounded}"
else:
return f"unknown config: {key}"
except ValueError as e:
return f"Error: invalid value - {e}"
except Exception as e:
logger.error(f"Set command error: {e}")
return f"Error: {e}"
# ==================== ACL Commands ====================
def _cmd_setperm(self, command: str) -> str:
"""Set permissions for a public key."""
# Format: setperm {pubkey-hex} {permissions-int}
parts = command[8:].split()
if len(parts) < 2:
return "Err - bad params"
pubkey_hex = parts[0]
try:
permissions = int(parts[1])
except ValueError:
return "Err - invalid permissions"
# TODO: Apply permissions via ACL
logger.info(f"setperm command: {pubkey_hex} -> {permissions}")
return "Error: Not yet implemented - use config file"
# ==================== Region Commands ====================
def _cmd_region(self, command: str) -> str:
"""Handle region commands."""
parts = command.split()
if len(parts) == 1:
return "Error: Region commands not implemented in Python repeater"
subcommand = parts[1]
if subcommand == "load":
return "Error: Region commands not implemented"
elif subcommand == "save":
return "Error: Region commands not implemented"
elif subcommand in ("allowf", "denyf", "get", "home", "put", "remove"):
return "Error: Region commands not implemented"
else:
return "Err - ??"
# ==================== Neighbor Commands ====================
def _cmd_neighbors(self) -> str:
"""List neighbors."""
# TODO: Get neighbors from routing table
return "Error: Not yet implemented"
def _cmd_neighbor_remove(self, command: str) -> str:
"""Remove a neighbor."""
pubkey_hex = command[16:].strip()
if not pubkey_hex:
return "ERR: Missing pubkey"
# TODO: Remove neighbor from routing table
logger.info(f"neighbor.remove: {pubkey_hex}")
return "Error: Not yet implemented"
# ==================== Temporary Radio Commands ====================
def _cmd_tempradio(self, command: str) -> str:
"""Apply temporary radio parameters."""
# Format: tempradio {freq} {bw} {sf} {cr} {timeout_mins}
parts = command[10:].split()
if len(parts) < 5:
return "Error: Expected freq bw sf cr timeout_mins"
try:
freq = float(parts[0])
bw = float(parts[1])
sf = int(parts[2])
cr = int(parts[3])
timeout_mins = int(parts[4])
# Validate
if not (300.0 <= freq <= 2500.0):
return "Error: invalid frequency"
if not (7.0 <= bw <= 500.0):
return "Error: invalid bandwidth"
if not (5 <= sf <= 12):
return "Error: invalid spreading factor"
if not (5 <= cr <= 8):
return "Error: invalid coding rate"
if timeout_mins <= 0:
return "Error: invalid timeout"
# TODO: Apply temporary radio parameters
logger.info(f"tempradio: {freq}MHz {bw}kHz SF{sf} CR4/{cr} for {timeout_mins}min")
return "Error: Not yet implemented"
except ValueError:
return "Error, invalid params"
# ==================== Logging Commands ====================
def _cmd_log(self, command: str) -> str:
"""Handle log commands."""
if command == "log start":
# TODO: Enable logging
return "Error: Not yet implemented"
elif command == "log stop":
# TODO: Disable logging
return "Error: Not yet implemented"
elif command == "log erase":
# TODO: Clear log file
return "Error: Not yet implemented"
elif command == "log":
return "Error: Use journalctl to view logs"
else:
return "Unknown log command"
# Backward compatibility alias
RepeaterCLI = MeshCLI
repeater/handler_helpers/text.py
+2
View File
@@ -18,6 +18,7 @@ from .room_server import RoomServer
logger = logging.getLogger("TextHelper")
# Text message type flags
TXT_TYPE_PLAIN = 0x00
TXT_TYPE_CLI_DATA = 0x01
@@ -464,6 +465,7 @@ class TextHelper:
"board",
"neighbors",
"neighbor.",
"discover.",
"tempradio ",
"setperm ",
"region",
repeater/main.py
+11 -8
View File
@@ -35,6 +35,7 @@ from repeater.handler_helpers import (
from repeater.identity_manager import IdentityManager
from repeater.packet_router import PacketRouter
from repeater.sensors import SensorManager
from repeater.utils_packet import create_scoped_advert_packet
from repeater.web.http_server import HTTPStatsServer, _log_buffer
logger = logging.getLogger("RepeaterDaemon")
@@ -1222,7 +1223,6 @@ class RepeaterDaemon:
return False
try:
from openhop_core.protocol import PacketBuilder
from openhop_core.protocol.constants import (
ADVERT_FLAG_HAS_NAME,
ADVERT_FLAG_IS_REPEATER,
@@ -1243,15 +1243,16 @@ class RepeaterDaemon:
flags = ADVERT_FLAG_IS_REPEATER | ADVERT_FLAG_HAS_NAME
packet = PacketBuilder.create_advert(
mesh_config = self.config.get("mesh", {})
default_region = mesh_config.get("default_region")
packet, scoped_region_name = create_scoped_advert_packet(
local_identity=self.local_identity,
name=node_name,
lat=latitude,
lon=longitude,
feature1=0,
feature2=0,
node_name=node_name,
latitude=latitude,
longitude=longitude,
flags=flags,
route_type="flood",
default_region=default_region,
scope_label="advert",
)
# Send via dispatcher
@@ -1270,6 +1271,8 @@ class RepeaterDaemon:
longitude,
location_source,
)
if scoped_region_name:
logger.info("Advert scoped to default region '%s'", scoped_region_name)
return True
except Exception as e:
repeater/utils_packet.py
+61
View File
@@ -0,0 +1,61 @@
import logging
from typing import Optional, Tuple
from openhop_core.protocol import PacketBuilder
from openhop_core.protocol.constants import ROUTE_TYPE_TRANSPORT_FLOOD
logger = logging.getLogger("RepeaterPacketUtils")
def create_scoped_advert_packet(
*,
local_identity,
node_name: str,
latitude: float,
longitude: float,
flags: int,
default_region,
scope_label: str,
) -> Tuple[object, Optional[str]]:
"""Create a flood advert packet and apply default-region transport scope when configured."""
packet = PacketBuilder.create_advert(
local_identity=local_identity,
name=node_name,
lat=latitude,
lon=longitude,
feature1=0,
feature2=0,
flags=flags,
route_type="flood",
)
scoped_region_name = _apply_default_region_scope(
packet=packet,
default_region=default_region,
scope_label=scope_label,
)
return packet, scoped_region_name
def _apply_default_region_scope(*, packet, default_region, scope_label: str) -> Optional[str]:
"""Apply transport-flood scoping for a default region if provided."""
region_name = str(default_region).strip() if default_region not in (None, "") else ""
if not region_name:
return None
try:
from openhop_core.protocol.transport_keys import calc_transport_code, get_auto_key_for
region_key = get_auto_key_for(region_name)
packet.transport_codes[0] = calc_transport_code(region_key, packet)
packet.transport_codes[1] = 0 # reserved for home region
packet.header = (packet.header & ~0x03) | ROUTE_TYPE_TRANSPORT_FLOOD
return region_name
except Exception as scope_err:
logger.warning(
"Failed to apply default region scope '%s' to %s; sending unscoped flood: %s",
region_name,
scope_label,
scope_err,
)
return None
repeater/web/api_endpoints.py
+162 -8
View File
@@ -28,6 +28,7 @@ from repeater.companion.utils import (
from repeater.config import resolve_storage_dir
from repeater.policy_engine import PolicyEngine
from repeater.service_utils import get_buildroot_image_info
from repeater.utils_packet import create_scoped_advert_packet
from .auth.middleware import require_auth
from .auth_endpoints import AuthAPIEndpoints
@@ -341,6 +342,8 @@ class APIEndpoints:
if not pub_key:
return enriched
enriched["is_self"] = self._is_local_discovery_pubkey(pub_key)
try:
enriched["node_hash"] = self._fmt_hash(bytes.fromhex(pub_key))
except ValueError:
@@ -381,6 +384,52 @@ class APIEndpoints:
enriched["advert_count"] = neighbor_info.get("advert_count")
return enriched
def _get_local_pubkey_hex(self) -> Optional[str]:
"""Return local node public key in hex when available."""
daemon = getattr(self, "daemon_instance", None)
identity = getattr(daemon, "local_identity", None)
try:
if identity and hasattr(identity, "get_public_key"):
pubkey = identity.get_public_key()
if isinstance(pubkey, (bytes, bytearray)):
return bytes(pubkey).hex().lower()
if isinstance(pubkey, str):
normalized = pubkey.strip().lower()
if normalized.startswith("0x"):
normalized = normalized[2:]
if normalized:
return normalized
except Exception as exc:
logger.debug("Unable to read local identity pubkey: %s", exc)
repeater_cfg = self.config.get("repeater", {}) if isinstance(self.config, dict) else {}
key = repeater_cfg.get("identity_key")
if isinstance(key, (bytes, bytearray)):
return bytes(key).hex().lower()
if isinstance(key, str):
normalized = key.strip().lower()
if normalized.startswith("0x"):
normalized = normalized[2:]
if normalized and all(ch in "0123456789abcdef" for ch in normalized):
return normalized
return None
def _is_local_discovery_pubkey(self, pub_key: str) -> bool:
"""Return True if discovery pub_key matches the local node key (including prefix form)."""
candidate = str(pub_key or "").strip().lower()
if not candidate:
return False
if candidate.startswith("0x"):
candidate = candidate[2:]
local_pubkey = self._get_local_pubkey_hex()
if not local_pubkey:
return False
return local_pubkey.startswith(candidate) or candidate.startswith(local_pubkey)
def _process_counter_data(self, data_points, timestamps_ms):
rates = []
prev_value = None
@@ -3323,6 +3372,7 @@ class APIEndpoints:
"direct_tx_delay_factor": 0.5, # Direct TX delay (0.0-5.0)
"rx_delay_base": 0.0, # RX delay base (>= 0)
"node_name": "MyNode", # Node name
"owner_info": "Owner text", # Owner info text
"latitude": 0.0, # Latitude (-90 to 90)
"longitude": 0.0, # Longitude (-180 to 180)
"max_flood_hops": 64, # Max flood hops (0-64)
@@ -3434,6 +3484,11 @@ class APIEndpoints:
self.config["repeater"]["node_name"] = name
applied.append(f"name={name}")
if "owner_info" in data:
owner_info = str(data["owner_info"]).replace("|", "\n")
self.config["repeater"]["owner_info"] = owner_info
applied.append("owner.info")
# Update latitude
if "latitude" in data:
lat = float(data["latitude"])
@@ -3950,6 +4005,90 @@ class APIEndpoints:
else:
return self._error("Method not supported")
@cherrypy.expose
@cherrypy.tools.json_out()
@cherrypy.tools.json_in()
def default_region(self):
"""
Get or update mesh default region configuration.
GET /default_region
POST /default_region
Body: {"default_region": "region-name" | null}
"""
if cherrypy.request.method == "GET":
try:
mesh_cfg = self.config.get("mesh", {}) if isinstance(self.config, dict) else {}
default_region = mesh_cfg.get("default_region")
value = str(default_region).strip() if default_region not in (None, "") else None
return self._success({"default_region": value})
except Exception as e:
logger.error(f"Error getting default region: {e}")
return self._error(e)
if cherrypy.request.method == "POST":
try:
data = cherrypy.request.json or {}
if "default_region" not in data:
return self._error("Missing required field: default_region")
raw_value = data.get("default_region")
default_region = None
if raw_value is not None:
text = str(raw_value).strip()
if text and text != "<null>":
default_region = text
if "mesh" not in self.config:
self.config["mesh"] = {}
# Keep region table compatible with firmware "region default": if non-null
# and missing, auto-create region and ensure flood allow.
if default_region:
storage = self._get_storage()
records = storage.get_transport_keys() or []
existing = None
needle = default_region.lower()
for rec in records:
name = str(rec.get("name") or "").strip()
display = name[1:] if name.startswith("#") else name
if display.lower() == needle:
existing = rec
break
if existing:
key_id = existing.get("id")
if key_id is not None:
storage.update_transport_key(int(key_id), flood_policy="allow")
else:
storage.create_transport_key(
default_region, "allow", None, None, time.time()
)
self.config["mesh"]["default_region"] = default_region
saved = self.config_manager.save_to_file()
if not saved:
return self._error("Failed to save configuration to file")
if hasattr(self.config_manager, "live_update_daemon"):
self.config_manager.live_update_daemon(["mesh"])
return self._success(
{"default_region": default_region},
message=(
"Default region cleared"
if default_region is None
else f"Default region set to {default_region}"
),
)
except Exception as e:
logger.error(f"Error updating default region: {e}")
return self._error(e)
return self._error("Method not supported")
@cherrypy.expose
@cherrypy.tools.json_out()
@cherrypy.tools.json_in()
@@ -4297,6 +4436,19 @@ class APIEndpoints:
if not re.fullmatch(r"[0-9a-f]{16}|[0-9a-f]{64}", pub_key):
return self._error("pub_key must be 8-byte or 32-byte hex")
if self._is_local_discovery_pubkey(pub_key):
enriched = self._enrich_discovery_result(
{
"pub_key": pub_key,
"node_name": node_name,
"node_type": node_type,
"rssi": rssi,
"response_snr": snr,
}
)
enriched["is_self"] = True
return self._success(enriched, message="Skipped local node: not added to neighbors")
contact_type = {
1: "Chat Node",
2: "Repeater",
@@ -5279,7 +5431,6 @@ class APIEndpoints:
):
"""Send advert for a room server identity"""
try:
from openhop_core.protocol import PacketBuilder
from openhop_core.protocol.constants import (
ADVERT_FLAG_HAS_NAME,
ADVERT_FLAG_IS_ROOM_SERVER,
@@ -5292,15 +5443,16 @@ class APIEndpoints:
# Build flags - just use HAS_NAME for room servers
flags = ADVERT_FLAG_IS_ROOM_SERVER | ADVERT_FLAG_HAS_NAME
packet = PacketBuilder.create_advert(
mesh_config = self.config.get("mesh", {}) if isinstance(self.config, dict) else {}
default_region = mesh_config.get("default_region")
packet, scoped_region_name = create_scoped_advert_packet(
local_identity=identity,
name=node_name,
lat=latitude,
lon=longitude,
feature1=0,
feature2=0,
node_name=node_name,
latitude=latitude,
longitude=longitude,
flags=flags,
route_type="flood",
default_region=default_region,
scope_label="room server advert",
)
# Send via dispatcher
@@ -5314,6 +5466,8 @@ class APIEndpoints:
logger.info(
f"Sent flood advert for room server '{node_name}' at ({latitude:.6f}, {longitude:.6f})"
)
if scoped_region_name:
logger.info("Room server advert scoped to default region '%s'", scoped_region_name)
return True
except Exception as e:
repeater/web/openapi.yaml
+55
View File
@@ -1541,6 +1541,61 @@ paths:
schema:
$ref: '#/components/schemas/SuccessResponse'
/default_region:
get:
tags: [Network Policy]
summary: Get default region
description: Get current mesh default region used for locally-originated flood adverts.
security:
- BearerAuth: []
- ApiKeyAuth: []
responses:
'200':
description: Current default region
content:
application/json:
schema:
type: object
properties:
success:
type: boolean
data:
type: object
properties:
default_region:
type: string
nullable: true
message:
type: string
error:
type: string
post:
tags: [Network Policy]
summary: Update default region
description: Set or clear mesh default region. Pass null to clear.
security:
- BearerAuth: []
- ApiKeyAuth: []
requestBody:
required: true
content:
application/json:
schema:
type: object
required: [default_region]
properties:
default_region:
type: string
nullable: true
description: Region name to set, or null to clear.
responses:
'200':
description: Default region updated
content:
application/json:
schema:
$ref: '#/components/schemas/SuccessResponse'
/ping_neighbor:
post:
tags: [Network Policy]
tests/test_api_endpoints_core_coverage.py
+126 -1
View File
@@ -1,7 +1,8 @@
import asyncio
from concurrent.futures import TimeoutError as FutureTimeoutError
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, mock_open, patch
from unittest.mock import AsyncMock, MagicMock, mock_open, patch
import cherrypy
import pytest
@@ -311,6 +312,43 @@ def test_add_discovered_neighbor_normalizes_unknown_name(cherrypy_ctx, monkeypat
assert advert_record["node_name"] is None
def test_add_discovered_neighbor_skips_local_node(cherrypy_ctx, monkeypatch):
request, _ = cherrypy_ctx
request.method = "POST"
request.headers = {"Authorization": "Bearer test-token"}
request.path_info = "/api/add_discovered_neighbor"
request.json = {
"pub_key": "aa" * 32,
"node_name": "Self",
"node_type": 2,
}
jwt_handler = MagicMock()
jwt_handler.verify_jwt.return_value = {"sub": "tester", "client_id": "ui"}
token_manager = MagicMock()
monkeypatch.setattr(
cherrypy,
"config",
{"jwt_handler": jwt_handler, "token_manager": token_manager},
raising=False,
)
storage = MagicMock()
api = _make_api()
api._enrich_discovery_result = APIEndpoints._enrich_discovery_result.__get__(api, APIEndpoints)
_attach_storage(api, storage)
api.daemon_instance.local_identity = SimpleNamespace(
get_public_key=lambda: bytes.fromhex("aa" * 32)
)
result = api.add_discovered_neighbor()
assert result["success"] is True
assert "Skipped local node" in result["message"]
assert result["data"]["is_self"] is True
storage.record_advert.assert_not_called()
def test_enrich_discovery_result_treats_placeholder_name_as_unknown():
api = _make_api()
@@ -1473,6 +1511,41 @@ def test_send_advert_paths(cherrypy_ctx):
future_timeout.cancel.assert_called_once()
def test_room_server_advert_applies_default_region_scope():
from openhop_core.protocol.constants import ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD
api = _make_api({"mesh": {"default_region": "alpha"}, "repeater": {}})
dispatcher = SimpleNamespace(send_packet=AsyncMock())
api.daemon_instance = SimpleNamespace(dispatcher=dispatcher, repeater_handler=None)
packet = SimpleNamespace(
header=ROUTE_TYPE_FLOOD,
transport_codes=[0, 0],
get_payload_type=lambda: 3,
get_payload=lambda: b"room_server_advert_payload",
)
with (
patch("openhop_core.protocol.PacketBuilder.create_advert", return_value=packet),
patch("openhop_core.protocol.transport_keys.get_auto_key_for", return_value=b"\x01" * 16),
patch("openhop_core.protocol.transport_keys.calc_transport_code", return_value=0xCAFE),
):
result = asyncio.run(
api._send_room_server_advert_async(
identity=SimpleNamespace(),
node_name="RoomAlpha",
latitude=1.0,
longitude=2.0,
disable_fwd=False,
)
)
assert result is True
assert packet.transport_codes == [0xCAFE, 0]
assert (packet.header & 0x03) == ROUTE_TYPE_TRANSPORT_FLOOD
dispatcher.send_packet.assert_awaited_once_with(packet, wait_for_ack=False)
def test_set_mode_and_set_duty_cycle_paths(cherrypy_ctx):
request, _ = cherrypy_ctx
api = _make_api({"repeater": {}, "duty_cycle": {}})
@@ -1757,13 +1830,40 @@ def test_transport_keys_and_transport_key_and_unscoped_policy(cherrypy_ctx):
request.method = "GET"
assert api.unscoped_flood_policy()["success"] is False
got_default = api.default_region()
assert got_default["success"] is True
assert got_default["data"]["default_region"] is None
request.method = "POST"
request.json = {}
assert api.unscoped_flood_policy()["success"] is False
request.json = {}
assert api.default_region()["success"] is False
request.json = {"unscoped_flood_allow": "yes"}
assert api.unscoped_flood_policy()["success"] is False
request.json = {"default_region": "alpha"}
set_default = api.default_region()
assert set_default["success"] is True
assert api.config["mesh"]["default_region"] == "alpha"
assert any(
call.args and call.args[0] == "alpha" and call.args[1] == "allow"
for call in storage.create_transport_key.call_args_list
)
request.method = "GET"
got_default_after = api.default_region()
assert got_default_after["success"] is True
assert got_default_after["data"]["default_region"] == "alpha"
request.method = "POST"
request.json = {"default_region": None}
clear_default = api.default_region()
assert clear_default["success"] is True
assert api.config["mesh"]["default_region"] is None
api.config_manager.save_to_file.return_value = True
request.json = {"unscoped_flood_allow": True}
ok = api.unscoped_flood_policy()
@@ -1771,6 +1871,31 @@ def test_transport_keys_and_transport_key_and_unscoped_policy(cherrypy_ctx):
assert api.config["mesh"]["unscoped_flood_allow"] is True
def test_update_radio_config_owner_info_and_mesh_fields(cherrypy_ctx):
request, _ = cherrypy_ctx
request.method = "POST"
request.json = {
"owner_info": "Alice|Ops",
"path_hash_mode": 2,
"loop_detect": "strict",
}
api = _make_api({"repeater": {}, "mesh": {}, "radio": {}, "delays": {}})
api.config_manager.update_and_save.return_value = {
"success": True,
"saved": True,
"live_updated": True,
}
out = api.update_radio_config()
assert out["success"] is True
assert api.config["repeater"]["owner_info"] == "Alice\nOps"
assert api.config["mesh"]["path_hash_mode"] == 2
assert api.config["mesh"]["loop_detect"] == "strict"
assert "owner.info" in out["data"].get("applied", [])
class _FakeIdentityObj:
def __init__(self, first=0x42):
self._pk = bytes([first]) + (b"A" * 31)
tests/test_companion_state_load.py
+3 -1
View File
@@ -30,7 +30,9 @@ class TestLoadCompanionRowsVerified:
async def test_genuinely_empty_returns_without_retry(self):
loader = MagicMock(return_value=[])
counter = MagicMock(return_value=0)
rows, stored = await _load_companion_rows_verified(loader, counter, "channels", _HASH, _NAME)
rows, stored = await _load_companion_rows_verified(
loader, counter, "channels", _HASH, _NAME
)
assert rows == []
assert stored == 0
assert loader.call_count == 1
tests/test_handler_helpers_mesh_cli.py
+200 -11
View File
@@ -1,7 +1,6 @@
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from repeater.handler_helpers.mesh_cli import MeshCLI
@@ -11,6 +10,7 @@ def _base_config():
"repeater": {
"name": "node-a",
"mode": "forward",
"owner_info": "owner-a",
"latitude": 1.2,
"longitude": 3.4,
"airtime_factor": 1.1,
@@ -31,6 +31,7 @@ def _base_config():
"coding_rate": 5,
"tx_power": 22,
},
"mesh": {"path_hash_mode": 0, "loop_detect": "minimal"},
"security": {"guest_password": "guest", "allow_read_only": True},
}
@@ -52,12 +53,13 @@ def test_handle_command_admin_and_prefix_behavior():
def test_help_routing_and_basic_unknown_paths():
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), enable_regions=False)
assert "pyMC CLI Commands" in cli._route_command("help")
assert "openHop CLI Commands" in cli._route_command("help")
assert "No detailed help" in cli._route_command("help nope")
assert cli._route_command("start ota").startswith("Error:")
assert cli._route_command("sensor read").startswith("Error:")
assert cli._route_command("gps on").startswith("Error:")
assert cli._route_command("stats-foo").startswith("Error:")
assert cli._route_command("discover.neighbors").startswith("Error:")
assert cli._route_command("region load x").startswith("Error: Region commands not available")
assert cli._route_command("unknown") == "Unknown command"
@@ -129,17 +131,24 @@ def test_cmd_get_public_key_and_neighbor_branches():
storage.get_neighbors = lambda: {
"aa": {"is_repeater": False, "zero_hop": False, "last_seen": 1}
}
assert "No repeaters or zero hop" in cli._cmd_neighbors()
assert "No zero hop repeaters" in cli._cmd_neighbors()
storage.get_neighbors = lambda: {
"abcdef12feed": {"is_repeater": True, "zero_hop": False, "last_seen": 10, "snr": 4.9},
"11223344aabb": {"is_repeater": False, "zero_hop": True, "last_seen": 20, "snr": 1.2},
"99aabbccddeeff00": {
"is_repeater": True,
"zero_hop": True,
"last_seen": 21,
"snr": 6.3,
},
}
with patch("time.time", return_value=30):
out = cli._cmd_neighbors()
assert "abcdef12:20:4" in out
assert "11223344:10:1" in out
assert "99aabbcc:9:6" in out
assert "abcdef12:20:4" not in out
assert "11223344:10:1" not in out
cli.storage_handler = SimpleNamespace(
get_neighbors=MagicMock(side_effect=RuntimeError("db fail"))
@@ -147,6 +156,14 @@ def test_cmd_get_public_key_and_neighbor_branches():
assert cli._cmd_neighbors().startswith("Error:")
def test_cmd_get_owner_info_path_hash_mode_and_loop_detect():
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr())
assert cli._cmd_get("owner.info") == "> owner-a"
assert cli._cmd_get("path.hash.mode") == "> 0"
assert cli._cmd_get("loop.detect") == "> minimal"
def test_cmd_set_updates_and_validation_errors():
cfg = _base_config()
mgr = _cfg_mgr()
@@ -167,7 +184,18 @@ def test_cmd_set_updates_and_validation_errors():
assert cli._cmd_set("freq 868000000").startswith("OK")
assert cli._cmd_set("tx 17") == "OK"
assert cli._cmd_set("guest.password g") == "OK"
assert cli._cmd_set("owner.info Alice|Ops") == "OK"
assert cfg["repeater"]["owner_info"] == "Alice\nOps"
assert cli._cmd_set("allow.read.only off") == "OK"
assert cli._cmd_set("path.hash.mode 2") == "OK"
assert cfg["mesh"]["path_hash_mode"] == 2
assert cli._cmd_set("path.hash.mode 3") == "Error: path.hash.mode must be 0, 1, or 2"
assert cli._cmd_set("loop.detect strict") == "OK"
assert cfg["mesh"]["loop_detect"] == "strict"
assert (
cli._cmd_set("loop.detect weird")
== "Error: loop.detect must be off, minimal, moderate, or strict"
)
assert cli._cmd_set("advert.interval 59").startswith("Error: interval range")
assert cli._cmd_set("flood.advert.interval 2").startswith("Error: interval range")
@@ -185,10 +213,12 @@ def test_cmd_set_updates_and_validation_errors():
def test_misc_commands_and_routes():
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), enable_regions=True)
assert cli._cmd_region("region").startswith("Error:")
assert cli._cmd_region("region load us").startswith("Error:")
assert cli._cmd_region("region save").startswith("Error:")
assert cli._cmd_region("region remove x").startswith("Error:")
cli.storage_handler = SimpleNamespace(get_transport_keys=lambda: [])
assert cli._cmd_region("region") == "* F"
assert cli._cmd_region("region load us").startswith("Err -")
assert cli._cmd_region("region save") == "OK"
assert cli._cmd_region("region remove x") == "Err - not found"
assert cli._cmd_region("region unknown").startswith("Err -")
assert cli._cmd_setperm("setperm") == "Err - bad params"
@@ -204,11 +234,170 @@ def test_misc_commands_and_routes():
assert cli._cmd_tempradio("tempradio 915 125 7 5 nope") == "Error, invalid params"
assert cli._cmd_tempradio("tempradio 915 125 7 5 10").startswith("Error:")
assert cli._cmd_neighbor_remove("neighbor.remove ") == "ERR: Missing pubkey"
assert cli._cmd_neighbor_remove("neighbor.remove abc").startswith("Error:")
assert cli._cmd_neighbor_remove("neighbor.remove") == "ERR: Missing pubkey"
storage = SimpleNamespace(delete_neighbors_by_pubkey_prefix=MagicMock(return_value=1))
cli.storage_handler = storage
assert cli._cmd_neighbor_remove("neighbor.remove abc") == "OK"
storage.delete_neighbors_by_pubkey_prefix.assert_called_with("abc")
assert cli._cmd_neighbor_remove("neighbor.remove ") == "OK"
storage.delete_neighbors_by_pubkey_prefix.assert_called_with(None)
assert cli._cmd_neighbor_remove("neighbor.remove zz") == "ERR: bad pubkey"
no_delete = SimpleNamespace()
cli.storage_handler = no_delete
assert "not supported" in cli._cmd_neighbor_remove("neighbor.remove ab")
cli.storage_handler = None
assert cli._cmd_neighbor_remove("neighbor.remove ab") == "Error: Storage not available"
discovery = SimpleNamespace(
cleanup_sessions=MagicMock(),
create_session=MagicMock(return_value={"session_id": "s1"}),
start_session_task=MagicMock(),
)
loop = SimpleNamespace(is_running=lambda: True, call_soon_threadsafe=MagicMock())
cli._event_loop = loop
cli.config_manager.daemon = SimpleNamespace(discovery_helper=discovery)
assert cli._cmd_discover_neighbors("discover.neighbors") == "OK - Discover sent"
discovery.cleanup_sessions.assert_called_once()
assert callable(discovery.create_session.call_args.kwargs["result_enricher"])
loop.call_soon_threadsafe.assert_called_once_with(discovery.start_session_task, "s1")
assert cli._cmd_discover_neighbors("discover.neighbors foo").startswith("Err -")
cli.config_manager.daemon = SimpleNamespace(discovery_helper=None)
assert cli._cmd_discover_neighbors("discover.neighbors").startswith("Error: Discovery helper")
assert cli._cmd_log("log start").startswith("Error:")
assert cli._cmd_log("log stop").startswith("Error:")
assert cli._cmd_log("log erase").startswith("Error:")
assert cli._cmd_log("log") == "Error: Use journalctl to view logs"
assert cli._cmd_log("log whatever") == "Unknown log command"
def test_region_command_response_shapes_and_ops():
keys = [
{
"id": 1,
"name": "alpha",
"flood_policy": "allow",
"parent_id": None,
},
{
"id": 2,
"name": "#beta",
"flood_policy": "deny",
"parent_id": 1,
},
]
created = []
updated = []
deleted = []
def get_transport_keys():
return list(keys)
def update_transport_key(key_id, **kwargs):
updated.append((key_id, kwargs))
for rec in keys:
if rec["id"] == key_id:
rec.update(kwargs)
return True
return False
def create_transport_key(name, flood_policy, transport_key, parent_id, last_used):
new_id = max([r["id"] for r in keys] + [0]) + 1
keys.append(
{
"id": new_id,
"name": name,
"flood_policy": flood_policy,
"parent_id": parent_id,
}
)
created.append((name, flood_policy, parent_id))
return new_id
def delete_transport_key(key_id):
for idx, rec in enumerate(keys):
if rec["id"] == key_id:
deleted.append(key_id)
del keys[idx]
return True
return False
cfg = _base_config()
cfg["mesh"] = {"unscoped_flood_allow": True, "default_region": None}
cli = MeshCLI("/tmp/cfg.yaml", cfg, _cfg_mgr(), enable_regions=True)
cli.storage_handler = SimpleNamespace(
get_transport_keys=get_transport_keys,
update_transport_key=update_transport_key,
create_transport_key=create_transport_key,
delete_transport_key=delete_transport_key,
)
tree = cli._cmd_region("region")
assert "* F" in tree
assert " alpha F" in tree
assert " beta" in tree
assert cli._cmd_region("region get alpha") == " alpha F"
assert cli._cmd_region("region get beta") == " beta (alpha)"
assert cli._cmd_region("region get missing") == "Err - unknown region"
assert cli._cmd_region("region allowf beta") == "OK"
assert keys[1]["flood_policy"] == "allow"
assert cli._cmd_region("region denyf beta") == "OK"
assert keys[1]["flood_policy"] == "deny"
assert cli._cmd_region("region home alpha") == " home is now alpha"
assert cli._cmd_region("region home") == " home is alpha"
assert cli._cmd_region("region default") == " default scope is <null>"
assert cli._cmd_region("region default alpha") == " default scope is now alpha"
assert cfg["mesh"]["default_region"] == "alpha"
assert cli._cmd_region("region default") == " default scope is alpha"
assert cli._cmd_region("region default <null>") == " default scope is now <null>"
assert cfg["mesh"]["default_region"] is None
assert cli._cmd_region("region put gamma alpha") == "OK - (flood allowed)"
assert any(r[0] == "gamma" for r in created)
assert cli._cmd_region("region remove alpha") == "Err - not empty"
assert cli._cmd_region("region remove gamma") == "OK"
assert deleted
assert cli._cmd_region("region list denied") == "beta"
listed = cli._cmd_region("region list allowed")
assert listed.startswith("*,")
assert "alpha" in listed
assert cli._cmd_region("region list nope") == "Err - use 'allowed' or 'denied'"
def test_discovery_auto_add_skips_local_node_and_persists_remote():
identity = SimpleNamespace(get_public_key=lambda: bytes.fromhex("11" * 32))
storage = SimpleNamespace(record_advert=MagicMock())
cli = MeshCLI(
"/tmp/cfg.yaml", _base_config(), _cfg_mgr(), identity=identity, storage_handler=storage
)
self_result = cli._auto_add_discovery_result({"pub_key": "11" * 8, "node_type": 2})
assert self_result["is_self"] is True
storage.record_advert.assert_not_called()
remote_result = cli._auto_add_discovery_result(
{
"pub_key": "22" * 32,
"node_name": "Remote Repeater",
"node_type": 2,
"rssi": -70,
"response_snr": 4.25,
}
)
assert remote_result["auto_added"] is True
storage.record_advert.assert_called_once()
tests/test_handler_helpers_path_protocol_text.py
+1
View File
@@ -312,6 +312,7 @@ def test_text_helper_cli_prefix_and_admin_permission_checks():
assert helper._is_cli_command("get status") is True
assert helper._is_cli_command("99|get status") is True
assert helper._is_cli_command("04|discover.neighbors") is True
assert helper._is_cli_command("hello world") is False
assert helper._check_admin_permission_for_identity(0x21, 0x41) is True
tests/test_handler_helpers_room_server.py
+4 -2
View File
@@ -6,9 +6,9 @@ import pytest
from repeater.handler_helpers.room_server import (
MAX_UNSYNCED_POSTS,
RoomServer,
TXT_TYPE_PLAIN,
TXT_TYPE_SIGNED_PLAIN,
RoomServer,
)
@@ -208,7 +208,9 @@ async def test_room_server_push_expected_ack_matches_firmware_signed_ack():
assert ok is False
assert len(sent) == 1
upserts = [c.kwargs for c in db.upsert_client_sync.call_args_list if "pending_ack_crc" in c.kwargs]
upserts = [
c.kwargs for c in db.upsert_client_sync.call_args_list if "pending_ack_crc" in c.kwargs
]
assert len(upserts) == 1
expected_ack_crc = upserts[0]["pending_ack_crc"]
tests/test_identity_manager.py
+58
View File
@@ -0,0 +1,58 @@
from repeater.identity_manager import IdentityManager
class _FakeIdentity:
def __init__(self, pubkey: bytes, addr: bytes = b"\xaa\xbb"):
self._pubkey = pubkey
self._addr = addr
def get_public_key(self):
return self._pubkey
def get_address_bytes(self):
return self._addr
def test_identity_manager_register_lookup_and_collision_paths():
mgr = IdentityManager(config={})
id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31, addr=b"\x01\x02")
id_b_collision = _FakeIdentity(bytes([0x11]) + b"B" * 31, addr=b"\x03\x04")
assert mgr.register_identity("alpha", id_a, {"k": 1}, "repeater") is True
assert mgr.has_identity(0x11) is True
assert mgr.get_identity_by_hash(0x11)[0] is id_a
assert mgr.get_identity_by_name("alpha")[0] is id_a
assert mgr.register_identity("beta", id_b_collision, {"k": 2}, "room_server") is False
def test_identity_manager_list_and_type_filtering():
mgr = IdentityManager(config={})
id_a = _FakeIdentity(bytes([0x22]) + b"A" * 31)
id_b = _FakeIdentity(bytes([0x33]) + b"B" * 31)
mgr.register_identity("rep-main", id_a, {"x": 1}, "repeater")
mgr.register_identity("room-a", id_b, {"y": 2}, "room_server")
listed = mgr.list_identities()
assert len(listed) == 2
assert any(item["hash"] == "0x22" and item["name"] == "repeater:rep-main" for item in listed)
assert any(item["hash"] == "0x33" and item["type"] == "room_server" for item in listed)
assert mgr.has_identity_type("repeater") is True
assert mgr.has_identity_type("room_server") is True
assert mgr.has_identity_type("unknown") is False
by_type = mgr.get_identities_by_type("room_server")
assert len(by_type) == 1
assert by_type[0][0] == "room-a"
def test_identity_manager_list_handles_none_identity_fields():
mgr = IdentityManager(config={})
mgr.identities[0x44] = (None, {}, "repeater")
mgr.registered_hashes[0x44] = "repeater:ghost"
listed = mgr.list_identities()
assert listed[0]["address"] == "N/A"
assert listed[0]["public_key"] is None
tests/test_identity_manager_and_repeater_cli.py
-267
View File
@@ -1,267 +0,0 @@
from unittest.mock import MagicMock, patch
from repeater.handler_helpers.repeater_cli import MeshCLI, RepeaterCLI
from repeater.identity_manager import IdentityManager
class _FakeIdentity:
def __init__(self, pubkey: bytes, addr: bytes = b"\xaa\xbb"):
self._pubkey = pubkey
self._addr = addr
def get_public_key(self):
return self._pubkey
def get_address_bytes(self):
return self._addr
def _base_config():
return {
"version": "9.9.9",
"repeater": {
"name": "node-1",
"mode": "forward",
"latitude": 12.3,
"longitude": 45.6,
"airtime_factor": 1.1,
"advert_interval_minutes": 120,
"flood_advert_interval_hours": 24,
"max_flood_hops": 32,
"rx_delay_base": 0.4,
"tx_delay_factor": 1.2,
"direct_tx_delay_factor": 0.7,
"multi_acks": 2,
"interference_threshold": -111,
"agc_reset_interval": 8,
},
"radio": {
"frequency": 915000000,
"bandwidth": 125000,
"spreading_factor": 7,
"coding_rate": 5,
"tx_power": 22,
},
"security": {"guest_password": "guest", "allow_read_only": True},
}
def test_identity_manager_register_lookup_and_collision_paths():
mgr = IdentityManager(config={})
id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31, addr=b"\x01\x02")
id_b_collision = _FakeIdentity(bytes([0x11]) + b"B" * 31, addr=b"\x03\x04")
assert mgr.register_identity("alpha", id_a, {"k": 1}, "repeater") is True
assert mgr.has_identity(0x11) is True
assert mgr.get_identity_by_hash(0x11)[0] is id_a
assert mgr.get_identity_by_name("alpha")[0] is id_a
# Collision on first pubkey byte should be rejected.
assert mgr.register_identity("beta", id_b_collision, {"k": 2}, "room_server") is False
def test_identity_manager_list_and_type_filtering():
mgr = IdentityManager(config={})
id_a = _FakeIdentity(bytes([0x22]) + b"A" * 31)
id_b = _FakeIdentity(bytes([0x33]) + b"B" * 31)
mgr.register_identity("rep-main", id_a, {"x": 1}, "repeater")
mgr.register_identity("room-a", id_b, {"y": 2}, "room_server")
listed = mgr.list_identities()
assert len(listed) == 2
assert any(item["hash"] == "0x22" and item["name"] == "repeater:rep-main" for item in listed)
assert any(item["hash"] == "0x33" and item["type"] == "room_server" for item in listed)
assert mgr.has_identity_type("repeater") is True
assert mgr.has_identity_type("room_server") is True
assert mgr.has_identity_type("unknown") is False
by_type = mgr.get_identities_by_type("room_server")
assert len(by_type) == 1
assert by_type[0][0] == "room-a"
def test_identity_manager_list_handles_none_identity_fields():
mgr = IdentityManager(config={})
mgr.identities[0x44] = (None, {}, "repeater")
mgr.registered_hashes[0x44] = "repeater:ghost"
listed = mgr.list_identities()
assert listed[0]["address"] == "N/A"
assert listed[0]["public_key"] is None
def test_repeater_cli_alias_points_to_mesh_cli():
assert RepeaterCLI is MeshCLI
def test_cli_non_admin_and_prefix_passthrough():
cfg = _base_config()
save = MagicMock()
cli = MeshCLI("/tmp/config.yaml", cfg, save)
assert cli.handle_command(b"x", "help", is_admin=False) == "Error: Admin permission required"
assert cli.handle_command(b"x", "01|help set", is_admin=True).startswith("01|")
def test_cli_help_and_route_unknown_commands():
cli = MeshCLI("/tmp/config.yaml", _base_config(), MagicMock())
help_text = cli._route_command("help")
assert "pyMC CLI Commands" in help_text
assert "No detailed help" in cli._route_command("help not-a-topic")
assert cli._route_command("start ota").startswith("Error:")
assert cli._route_command("gps now").startswith("Error:")
assert cli._route_command("stats-air").startswith("Error:")
assert cli._route_command("totally-unknown") == "Unknown command"
def test_cli_reboot_uses_service_utils_result():
cli = MeshCLI("/tmp/config.yaml", _base_config(), MagicMock())
with patch("repeater.service_utils.restart_service", return_value=(True, "restarted")):
assert cli._cmd_reboot() == "OK - restarted"
with patch("repeater.service_utils.restart_service", return_value=(False, "denied")):
assert cli._cmd_reboot() == "Error: denied"
def test_cli_clock_time_password_and_version_commands():
cfg = _base_config()
save = MagicMock()
cli = MeshCLI("/tmp/config.yaml", cfg, save, identity_type="room_server")
assert "UTC" in cli._cmd_clock("clock")
assert "not needed" in cli._cmd_clock("clock sync")
assert cli._cmd_clock("clock bad") == "Unknown clock command"
assert cli._cmd_time("time 1 2").startswith("Error:")
assert cli._cmd_password("password ") == "Error: Password cannot be empty"
assert cli._cmd_password("password newpass") == "password now: newpass"
assert cfg["security"]["password"] == "newpass"
save.assert_called()
assert cli._cmd_version() == "pyMC_room_server v9.9.9"
def test_cli_get_commands_cover_expected_fields():
cli = MeshCLI("/tmp/config.yaml", _base_config(), MagicMock())
assert cli._cmd_get("af") == "> 1.1"
assert cli._cmd_get("name") == "> node-1"
assert cli._cmd_get("repeat") == "> on"
assert cli._cmd_get("lat") == "> 12.3"
assert cli._cmd_get("lon") == "> 45.6"
assert cli._cmd_get("radio") == "> 915.0,125.0,7,5"
assert cli._cmd_get("freq") == "> 915.0"
assert cli._cmd_get("tx") == "> 22"
assert cli._cmd_get("role") == "> repeater"
assert cli._cmd_get("guest.password") == "> guest"
assert cli._cmd_get("allow.read.only") == "> on"
assert cli._cmd_get("advert.interval") == "> 120"
assert cli._cmd_get("flood.advert.interval") == "> 24"
assert cli._cmd_get("flood.max") == "> 32"
assert cli._cmd_get("rxdelay") == "> 0.4"
assert cli._cmd_get("txdelay") == "> 1.2"
assert cli._cmd_get("direct.txdelay") == "> 0.7"
assert cli._cmd_get("multi.acks") == "> 2"
assert cli._cmd_get("int.thresh") == "> -111"
assert cli._cmd_get("agc.reset.interval") == "> 8"
assert cli._cmd_get("public.key").startswith("Error:")
assert cli._cmd_get("missing") == "??: missing"
def test_cli_set_commands_apply_and_validate_ranges():
cfg = _base_config()
save = MagicMock()
cli = MeshCLI("/tmp/config.yaml", cfg, save)
assert cli._cmd_set("af 2.5") == "OK"
assert cfg["repeater"]["airtime_factor"] == 2.5
assert cli._cmd_set("name repeater-z") == "OK"
assert cfg["repeater"]["name"] == "repeater-z"
assert cli._cmd_set("repeat off").endswith("OFF")
assert cfg["repeater"]["mode"] == "monitor"
assert cli._cmd_set("lat 1.25") == "OK"
assert cli._cmd_set("lon 2.5") == "OK"
assert cli._cmd_set("radio 900000000 250000 9 6").startswith("OK")
assert cfg["radio"]["frequency"] == 900000000.0
assert cli._cmd_set("freq 868000000").startswith("OK")
assert cli._cmd_set("tx 17") == "OK"
assert cli._cmd_set("guest.password gpw") == "OK"
assert cli._cmd_set("allow.read.only off") == "OK"
assert cli._cmd_set("advert.interval 59").startswith("Error: interval range")
assert cli._cmd_set("advert.interval 60") == "OK"
assert cli._cmd_set("flood.advert.interval 2").startswith("Error: interval range")
assert cli._cmd_set("flood.advert.interval 168") == "OK"
assert cli._cmd_set("flood.max 65") == "Error: max 64"
assert cli._cmd_set("flood.max 64") == "OK"
assert cli._cmd_set("rxdelay -1") == "Error: cannot be negative"
assert cli._cmd_set("txdelay -1") == "Error: cannot be negative"
assert cli._cmd_set("direct.txdelay -1") == "Error: cannot be negative"
assert cli._cmd_set("multi.acks 5") == "OK"
assert cli._cmd_set("int.thresh -120") == "OK"
assert cli._cmd_set("agc.reset.interval 10") == "OK - interval rounded to 8"
def test_cli_set_command_error_paths():
cfg = _base_config()
save = MagicMock()
cli = MeshCLI("/tmp/config.yaml", cfg, save)
assert cli._cmd_set("af") == "Error: Missing value"
assert cli._cmd_set("radio 1 2 3") == "Error: Expected freq bw sf cr"
assert cli._cmd_set("unknown.key 1") == "unknown config: unknown.key"
assert cli._cmd_set("tx not-int").startswith("Error: invalid value")
cli.save_config = MagicMock(side_effect=RuntimeError("disk full"))
assert cli._cmd_set("name x").startswith("Error:")
def test_cli_setperm_region_neighbor_tempradio_log_paths():
cli = MeshCLI("/tmp/config.yaml", _base_config(), MagicMock(), enable_regions=False)
assert cli._cmd_setperm("setperm") == "Err - bad params"
assert cli._cmd_setperm("setperm deadbeef zz") == "Err - invalid permissions"
assert cli._cmd_setperm("setperm deadbeef 2").startswith("Error:")
assert "not available" in cli._route_command("region load us")
cli_regions = MeshCLI("/tmp/config.yaml", _base_config(), MagicMock(), enable_regions=True)
assert cli_regions._cmd_region("region").startswith("Error:")
assert cli_regions._cmd_region("region load x").startswith("Error:")
assert cli_regions._cmd_region("region save").startswith("Error:")
assert cli_regions._cmd_region("region allowf").startswith("Error:")
assert cli_regions._cmd_region("region what").startswith("Err -")
assert cli._cmd_neighbors().startswith("Error:")
assert cli._cmd_neighbor_remove("neighbor.remove ") == "ERR: Missing pubkey"
assert cli._cmd_neighbor_remove("neighbor.remove 001122").startswith("Error:")
assert cli._cmd_tempradio("tempradio 1 2 3").startswith("Error:")
assert cli._cmd_tempradio("tempradio 299 125 7 5 10") == "Error: invalid frequency"
assert cli._cmd_tempradio("tempradio 915 6 7 5 10") == "Error: invalid bandwidth"
assert cli._cmd_tempradio("tempradio 915 125 4 5 10") == "Error: invalid spreading factor"
assert cli._cmd_tempradio("tempradio 915 125 7 9 10") == "Error: invalid coding rate"
assert cli._cmd_tempradio("tempradio 915 125 7 5 0") == "Error: invalid timeout"
assert cli._cmd_tempradio("tempradio 915 125 7 5 x") == "Error, invalid params"
assert cli._cmd_tempradio("tempradio 915 125 7 5 10").startswith("Error:")
assert cli._cmd_log("log start").startswith("Error:")
assert cli._cmd_log("log stop").startswith("Error:")
assert cli._cmd_log("log erase").startswith("Error:")
assert cli._cmd_log("log") == "Error: Use journalctl to view logs"
assert cli._cmd_log("log weird") == "Unknown log command"
tests/test_main_py_coverage.py
+33
View File
@@ -264,6 +264,39 @@ async def test_send_advert_branches_and_success_path():
daemon.dispatcher.packet_filter.track_packet.assert_called_once()
@pytest.mark.asyncio
async def test_send_advert_applies_transport_scope_when_default_region_set():
from openhop_core.protocol.constants import ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD
daemon = RepeaterDaemon(_base_config(), radio=object())
daemon.dispatcher = SimpleNamespace(
send_packet=AsyncMock(), packet_filter=SimpleNamespace(track_packet=MagicMock())
)
daemon.local_identity = _FakeIdentity(b"\x22" + b"y" * 31)
daemon.repeater_handler = SimpleNamespace(mark_seen=MagicMock())
daemon.config["mesh"] = {"default_region": "alpha"}
packet = SimpleNamespace(
header=ROUTE_TYPE_FLOOD,
transport_codes=[0, 0],
get_payload_type=lambda: 3,
get_payload=lambda: b"minimal_advert_payload",
calculate_packet_hash=lambda: b"\xcd" * 16,
)
with (
patch("openhop_core.protocol.PacketBuilder.create_advert", return_value=packet),
patch("openhop_core.protocol.transport_keys.get_auto_key_for", return_value=b"\x01" * 16),
patch("openhop_core.protocol.transport_keys.calc_transport_code", return_value=0xBEEF),
):
ok = await daemon.send_advert()
assert ok is True
assert packet.transport_codes == [0xBEEF, 0]
assert (packet.header & 0x03) == ROUTE_TYPE_TRANSPORT_FLOOD
daemon.dispatcher.send_packet.assert_awaited_once_with(packet, wait_for_ack=False)
def test_update_repeater_location_from_gps_branches():
daemon = RepeaterDaemon(_base_config(), radio=object())