From 14b4804c267f6f8142ac62f260655191c439274d Mon Sep 17 00:00:00 2001 From: Rightup Date: Thu, 4 Jun 2026 15:53:17 +0100 Subject: [PATCH] feat: Enhance logging system and introduce policy management endpoints - Updated LogBuffer to support log entry IDs, enhanced log entry structure with additional metadata, and implemented subscriber management for real-time log streaming. - Added OpenAPI specifications for new endpoints related to policy management, including retrieval, updating, validation, and group management for network policies. - Implemented comprehensive tests for new policy endpoints, ensuring correct behavior for creating, updating, validating, and deleting policy groups and entries. - Introduced policy evaluation tests to validate the functionality of the PolicyEngine, including various scenarios for action decisions based on defined rules. - Enhanced packet routing tests to ensure proper handling of policy decisions in packet processing. --- .pre-commit-config.yaml | 5 +- config.yaml.example | 5 + repeater/config.py | 60 ++ repeater/data_acquisition/sqlite_handler.py | 41 + .../data_acquisition/storage_collector.py | 12 + repeater/engine.py | 44 +- repeater/packet_router.py | 105 ++- repeater/policy_engine.py | 677 ++++++++++++++++ repeater/web/api_endpoints.py | 753 +++++++++++++++++- repeater/web/http_server.py | 73 +- repeater/web/openapi.yaml | 289 +++++++ tests/test_api_endpoints_core_coverage.py | 36 + tests/test_api_policy_endpoints.py | 253 ++++++ tests/test_packet_router.py | 99 +++ tests/test_policy_engine.py | 749 +++++++++++++++++ 15 files changed, 3160 insertions(+), 41 deletions(-) create mode 100644 repeater/policy_engine.py create mode 100644 tests/test_api_policy_endpoints.py create mode 100644 tests/test_policy_engine.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index bd1c5ab..3c45797 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -46,8 +46,9 @@ repos: hooks: - id: openapi-contract-check name: OpenAPI contract check - entry: python3 scripts/check_openapi_contract.py - language: system + entry: python scripts/check_openapi_contract.py + language: python + additional_dependencies: [PyYAML] pass_filenames: false always_run: true files: ^(repeater/web/.*\.py|repeater/web/openapi\.yaml)$ diff --git a/config.yaml.example b/config.yaml.example index e7888ec..504aa81 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -122,6 +122,11 @@ repeater: # Controls how long users stay logged in before needing to re-authenticate jwt_expiry_minutes: 60 +# Policy engine configuration file. +# Relative paths are resolved from this config file directory. +policy: + policy_file: "policy.yaml" + # Local GPS receiver. When enabled, the daemon reads NMEA sentences from the # configured source and exposes parsed data at /api/gps. gps: diff --git a/repeater/config.py b/repeater/config.py index 1fc2f74..541235d 100644 --- a/repeater/config.py +++ b/repeater/config.py @@ -6,9 +6,67 @@ from typing import Any, Dict, Optional, overload import yaml +from repeater.policy_engine import default_policy_engine_config + logger = logging.getLogger("Config") +def _resolve_policy_config_path(config: Dict[str, Any], config_path: str) -> Path: + policy_section = config.get("policy", {}) if isinstance(config.get("policy"), dict) else {} + configured = policy_section.get("policy_file") or "policy.yaml" + + base_dir = Path(config_path).expanduser().resolve().parent + p = Path(str(configured)).expanduser() + if not p.is_absolute(): + p = (base_dir / p).resolve() + return p + + +def _load_policy_engine_config(config: Dict[str, Any], config_path: str) -> Dict[str, Any]: + policy_path = _resolve_policy_config_path(config, config_path) + defaults = default_policy_engine_config() + + if not policy_path.exists(): + logger.info("Policy file not found at %s, policy engine disabled", policy_path) + config["policy_engine"] = defaults + config["policy_file_path"] = str(policy_path) + return config + + try: + with open(policy_path) as f: + loaded = yaml.safe_load(f) or {} + + if isinstance(loaded, dict) and isinstance(loaded.get("policy_engine"), dict): + policy_cfg = loaded.get("policy_engine") + elif isinstance(loaded, dict): + policy_cfg = loaded + else: + policy_cfg = {} + + merged = dict(defaults) + if isinstance(policy_cfg, dict): + merged.update(policy_cfg) + + if not isinstance(merged.get("rules"), list): + merged["rules"] = [] + if not isinstance(merged.get("objects"), dict): + merged["objects"] = {} + + config["policy_engine"] = merged + config["policy_file_path"] = str(policy_path) + logger.info("Loaded policy config from %s", policy_path) + return config + except Exception as e: + logger.warning( + "Failed to load policy config from %s: %s. Policy engine disabled.", + policy_path, + e, + ) + config["policy_engine"] = defaults + config["policy_file_path"] = str(policy_path) + return config + + class NullRadio: """No-op radio used when radio_type disables hardware initialization.""" @@ -207,6 +265,8 @@ def load_config(config_path: Optional[str] = None) -> Dict[str, Any]: config["logging"] = {} config["logging"]["level"] = os.getenv("PYMC_REPEATER_LOG_LEVEL") + config = _load_policy_engine_config(config, config_path) + return config diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py index fc15edc..9ab18cb 100644 --- a/repeater/data_acquisition/sqlite_handler.py +++ b/repeater/data_acquisition/sqlite_handler.py @@ -860,6 +860,47 @@ class SQLiteHandler: logger.error(f"Failed to get CRC error history: {e}") return [] + def get_policy_event_counts( + self, + start_timestamp: float, + end_timestamp: float, + bucket_seconds: int = 60, + ) -> list: + """Return policy-blocked packet counts grouped by bucket timestamp. + + A policy event is represented by a packet drop reason that starts with + "Policy blocked packet". + """ + try: + bucket_seconds = max(1, int(bucket_seconds)) + with self._connect() as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + """ + SELECT + CAST(timestamp / ? AS INTEGER) * ? AS bucket_ts, + COUNT(*) AS count + FROM packets + WHERE timestamp >= ? + AND timestamp <= ? + AND drop_reason LIKE 'Policy blocked packet%' + GROUP BY bucket_ts + ORDER BY bucket_ts ASC + """, + (bucket_seconds, bucket_seconds, start_timestamp, end_timestamp), + ).fetchall() + + return [ + { + "timestamp": int(row["bucket_ts"]), + "count": int(row["count"]), + } + for row in rows + ] + except Exception as e: + logger.error(f"Failed to get policy event counts: {e}") + return [] + def get_packet_stats(self, hours: int = 24) -> dict: try: now = time.time() diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index fb363d7..f004e0d 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -310,6 +310,18 @@ class StorageCollector: def get_crc_error_history(self, hours: int = 24, limit: int = None) -> list: return self.sqlite_handler.get_crc_error_history(hours, limit) + def get_policy_event_counts( + self, + start_timestamp: float, + end_timestamp: float, + bucket_seconds: int = 60, + ) -> list: + return self.sqlite_handler.get_policy_event_counts( + start_timestamp=start_timestamp, + end_timestamp=end_timestamp, + bucket_seconds=bucket_seconds, + ) + def get_packet_stats(self, hours: int = 24) -> dict: return self.sqlite_handler.get_packet_stats(hours) diff --git a/repeater/engine.py b/repeater/engine.py index a909ac4..f0c1abc 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -23,6 +23,7 @@ from pymc_core.protocol.packet_utils import PacketHeaderUtils, PathUtils from repeater.airtime import AirtimeManager from repeater.data_acquisition import StorageCollector +from repeater.policy_engine import PolicyDecision, PolicyEngine logger = logging.getLogger("RepeaterHandler") @@ -65,6 +66,7 @@ class RepeaterHandler(BaseHandler): self.local_hash_bytes = local_hash_bytes or bytes([local_hash]) self.send_advert_func = send_advert_func self.airtime_mgr = AirtimeManager(config) + self.policy_engine = PolicyEngine.from_runtime_config(config) self.seen_packets = OrderedDict() self.cache_ttl = max( 300, config.get("repeater", {}).get("cache_ttl", 3600) @@ -191,6 +193,38 @@ class RepeaterHandler(BaseHandler): allow_forward = mode == "forward" allow_local_tx = mode != "no_tx" + policy_context = { + "route_type": route_type, + "payload_type": packet.get_payload_type() + if hasattr(packet, "get_payload_type") + else None, + "payload_length": len(packet.payload or b""), + "path_hash_size": packet.get_path_hash_size() + if hasattr(packet, "get_path_hash_size") + else None, + "hop_count": packet.get_path_hash_count() + if hasattr(packet, "get_path_hash_count") + else None, + "rssi": metadata.get("rssi", 0), + "snr": metadata.get("snr", 0.0), + "local_transmission": local_transmission, + "mode": mode, + } + prechecked_decision = metadata.get("_policy_precheck_decision") + if isinstance(prechecked_decision, PolicyDecision): + policy_decision = prechecked_decision + else: + policy_decision = self.policy_engine.evaluate(packet, policy_context) + policy_reason = None + + if policy_decision.matched: + logger.info(policy_decision.reason) + + if policy_decision.action == "drop": + allow_forward = False + allow_local_tx = False + policy_reason = self._policy_drop_reason(policy_decision) + if logger.isEnabledFor(logging.DEBUG): logger.debug( f"RX packet: header=0x{packet.header:02x}, payload_len={len(packet.payload or b'')}, " @@ -329,9 +363,9 @@ class RepeaterHandler(BaseHandler): self.dropped_count += 1 # Determine drop reason if local_transmission and not allow_local_tx: - drop_reason = "No TX mode" + drop_reason = policy_reason or "No TX mode" elif not allow_forward: - drop_reason = "Repeat disabled" + drop_reason = policy_reason or "Repeat disabled" else: # Check if packet has a specific drop reason set by handlers drop_reason = processed_packet.drop_reason or self._get_drop_reason( @@ -425,6 +459,12 @@ class RepeaterHandler(BaseHandler): return transmitted + @staticmethod + def _policy_drop_reason(decision: PolicyDecision) -> str: + if decision.rule_id is None: + return "Policy blocked packet" + return f"Policy blocked packet (rule {decision.rule_id})" + def log_trace_record(self, packet_record: dict) -> None: """Manually log a packet trace record (used by external callers)""" self._append_recent_packet(packet_record) diff --git a/repeater/packet_router.py b/repeater/packet_router.py index 63da5de..13472ca 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -19,6 +19,8 @@ from pymc_core.protocol.constants import ( ROUTE_TYPE_TRANSPORT_DIRECT, ) +from repeater.policy_engine import PolicyDecision, PolicyEngine + logger = logging.getLogger("PacketRouter") # Deliver PATH and protocol-response (PATH) to companion at most once per logical packet @@ -188,6 +190,68 @@ class PacketRouter: self._companion_delivered[key] = now + _COMPANION_DEDUPE_TTL_SEC return True + def _policy_companion_decision(self, packet, metadata: dict) -> PolicyDecision | None: + """Return cached policy decision used to gate companion delivery. + + Stores the pre-check decision in shared metadata so the repeater engine + can reuse it and avoid a second full policy evaluation pass. + """ + handler = getattr(self.daemon, "repeater_handler", None) + if not handler: + return None + policy_engine = getattr(handler, "policy_engine", None) + if not isinstance(policy_engine, PolicyEngine) or not policy_engine.enabled: + return None + + cached = metadata.get("_policy_precheck_decision") + if isinstance(cached, PolicyDecision): + return cached + + mode = self.daemon.config.get("repeater", {}).get("mode", "forward") + route_type = getattr(packet, "header", 0) & PH_ROUTE_MASK + policy_context = { + "route_type": route_type, + "payload_type": packet.get_payload_type() + if hasattr(packet, "get_payload_type") + else None, + "payload_length": len(packet.payload or b""), + "path_hash_size": packet.get_path_hash_size() + if hasattr(packet, "get_path_hash_size") + else None, + "hop_count": packet.get_path_hash_count() + if hasattr(packet, "get_path_hash_count") + else None, + "rssi": metadata.get("rssi", getattr(packet, "rssi", 0)), + "snr": metadata.get("snr", getattr(packet, "snr", 0.0)), + "local_transmission": False, + "mode": mode, + } + decision = policy_engine.evaluate(packet, policy_context) + metadata["_policy_precheck_decision"] = decision + return decision + + def _policy_blocks_companion(self, packet, metadata: dict) -> bool: + """Return True when policy action is drop, making companion suppression final.""" + decision = self._policy_companion_decision(packet, metadata) + if not isinstance(decision, PolicyDecision): + return False + if decision.action == "drop": + logger.debug( + "Policy pre-check blocked companion delivery: rule %s action=drop", + decision.rule_id, + ) + return True + return False + + def _companion_bridges_for_packet(self, packet, metadata: dict) -> dict: + """Return companion bridges unless policy drop pre-check blocks delivery.""" + companion_bridges = getattr(self.daemon, "companion_bridges", {}) + if not companion_bridges: + return {} + if self._policy_blocks_companion(packet, metadata): + return {} + return companion_bridges + def _record_for_ui(self, packet, metadata: dict) -> None: """Record an injection-only packet for the web UI (storage + recent_packets).""" handler = getattr(self.daemon, "repeater_handler", None) @@ -373,8 +437,10 @@ class PacketRouter: rssi = getattr(packet, "rssi", 0) snr = getattr(packet, "snr", 0.0) await self.daemon.advert_helper.process_advert_packet(packet, rssi, snr) - # Also feed adverts to companion bridges (for contact/path updates) - for bridge in getattr(self.daemon, "companion_bridges", {}).values(): + # Also feed adverts to companion bridges (for contact/path updates), + # but keep policy drop final just like the other companion paths. + companion_bridges = self._companion_bridges_for_packet(packet, metadata) + for bridge in companion_bridges.values(): try: await bridge.process_received_packet(packet) except Exception as e: @@ -385,7 +451,7 @@ class PacketRouter: # When dest is remote (not handled), pass to engine so DIRECT/FLOOD ANON_REQ can be forwarded. # Our own injected ANON_REQ is suppressed by the engine's duplicate (mark_seen) check. dest_hash = packet.payload[0] if packet.payload else None - companion_bridges = getattr(self.daemon, "companion_bridges", {}) + companion_bridges = self._companion_bridges_for_packet(packet, metadata) if dest_hash is not None and dest_hash in companion_bridges: await companion_bridges[dest_hash].process_received_packet(packet) processed_by_injection = True @@ -399,7 +465,7 @@ class PacketRouter: elif payload_type == AckHandler.payload_type(): # ACK has no dest in payload (4-byte CRC only); deliver to all bridges so sender sees send_confirmed. # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop. - companion_bridges = getattr(self.daemon, "companion_bridges", {}) + companion_bridges = self._companion_bridges_for_packet(packet, metadata) for bridge in companion_bridges.values(): try: await bridge.process_received_packet(packet) @@ -408,7 +474,7 @@ class PacketRouter: elif payload_type == TextMessageHandler.payload_type(): dest_hash = packet.payload[0] if packet.payload else None - companion_bridges = getattr(self.daemon, "companion_bridges", {}) + companion_bridges = self._companion_bridges_for_packet(packet, metadata) if dest_hash is not None and dest_hash in companion_bridges: await companion_bridges[dest_hash].process_received_packet(packet) processed_by_injection = True @@ -421,7 +487,7 @@ class PacketRouter: elif payload_type == PathHandler.payload_type(): dest_hash = packet.payload[0] if packet.payload else None - companion_bridges = getattr(self.daemon, "companion_bridges", {}) + companion_bridges = self._companion_bridges_for_packet(packet, metadata) if dest_hash is not None and dest_hash in companion_bridges: if self._should_deliver_path_to_companions(packet): await companion_bridges[dest_hash].process_received_packet(packet) @@ -450,7 +516,7 @@ class PacketRouter: # to first hop instead of original requester). # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop. dest_hash = packet.payload[0] if packet.payload and len(packet.payload) >= 1 else None - companion_bridges = getattr(self.daemon, "companion_bridges", {}) + companion_bridges = self._companion_bridges_for_packet(packet, metadata) local_hash = getattr(self.daemon, "local_hash", None) if dest_hash is not None and dest_hash in companion_bridges: try: @@ -496,7 +562,7 @@ class PacketRouter: # PAYLOAD_TYPE_PATH (0x08): protocol responses (telemetry, binary, etc.). # Deliver at most once per logical packet so the client is not spammed with duplicates. # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop. - companion_bridges = getattr(self.daemon, "companion_bridges", {}) + companion_bridges = self._companion_bridges_for_packet(packet, metadata) if companion_bridges and self._should_deliver_path_to_companions(packet): for bridge in companion_bridges.values(): try: @@ -516,7 +582,7 @@ class PacketRouter: elif payload_type == ProtocolRequestHandler.payload_type(): dest_hash = packet.payload[0] if packet.payload else None - companion_bridges = getattr(self.daemon, "companion_bridges", {}) + companion_bridges = self._companion_bridges_for_packet(packet, metadata) if dest_hash is not None and dest_hash in companion_bridges: await companion_bridges[dest_hash].process_received_packet(packet) processed_by_injection = True @@ -537,24 +603,21 @@ class PacketRouter: self._record_for_ui(packet, metadata) elif payload_type == GroupTextHandler.payload_type(): - # GRP_TXT: pass to all companions (they filter by channel); still forward - companion_bridges = getattr(self.daemon, "companion_bridges", {}) - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge GRP_TXT error: {e}") + # GRP_TXT: pass to all companions (they filter by channel); still forward. + # Policy drop is final and blocks companion delivery. + companion_bridges = self._companion_bridges_for_packet(packet, metadata) + if companion_bridges: + for bridge in companion_bridges.values(): + try: + await bridge.process_received_packet(packet) + except Exception as e: + logger.debug(f"Companion bridge GRP_TXT error: {e}") # Only pass to repeater engine if not already processed by injection # Skip engine for packets we injected for TX (already sent; avoid double-send/double-count) if getattr(packet, "_injected_for_tx", False): processed_by_injection = True if self.daemon.repeater_handler and not processed_by_injection: - metadata = { - "rssi": getattr(packet, "rssi", 0), - "snr": getattr(packet, "snr", 0.0), - "timestamp": getattr(packet, "timestamp", 0), - } sent = await self.daemon.repeater_handler(packet, metadata) if sent is False: drop_reason = getattr(packet, "_repeater_drop_reason", None) diff --git a/repeater/policy_engine.py b/repeater/policy_engine.py new file mode 100644 index 0000000..1573e72 --- /dev/null +++ b/repeater/policy_engine.py @@ -0,0 +1,677 @@ +from __future__ import annotations + +import hashlib +import logging +from dataclasses import dataclass +from typing import Any, Optional + +from pymc_core.protocol.constants import PAYLOAD_TYPE_GRP_DATA, PAYLOAD_TYPE_GRP_TXT +from pymc_core.protocol.crypto import CryptoUtils + +logger = logging.getLogger("PolicyEngine") + + +SUPPORTED_ACTIONS = { + "allow", + "drop", + "log_only", +} + + +def default_policy_engine_config() -> dict[str, Any]: + return { + "enabled": False, + "default_action": "allow", + "rules": [], + "objects": {}, + } + + +@dataclass +class PolicyDecision: + action: str = "allow" + matched: bool = False + rule_id: Optional[Any] = None + reason: Optional[str] = None + + +class PolicyEngine: + """Readable top-down rule evaluator for repeater policy decisions.""" + + def __init__(self, policy_config: Optional[dict] = None): + cfg = default_policy_engine_config() + if isinstance(policy_config, dict): + cfg.update(policy_config) + + self.enabled = bool(cfg.get("enabled", False)) + self.default_action = str(cfg.get("default_action", "allow")) + if self.default_action not in SUPPORTED_ACTIONS: + logger.warning( + "Policy default_action '%s' is not supported, using 'allow'", + self.default_action, + ) + self.default_action = "allow" + + self.rules = cfg.get("rules") if isinstance(cfg.get("rules"), list) else [] + self.objects = cfg.get("objects") if isinstance(cfg.get("objects"), dict) else {} + self._channel_decrypt_cache: dict[int, dict[str, Any]] = {} + self._inline_channel_secrets = self._collect_inline_rule_channel_secrets(self.rules) + + @classmethod + def from_runtime_config(cls, runtime_config: Optional[dict]) -> "PolicyEngine": + if not isinstance(runtime_config, dict): + return cls() + return cls(runtime_config.get("policy_engine", {})) + + def evaluate(self, packet, context: dict) -> PolicyDecision: + if not self.enabled: + return PolicyDecision(action="allow", matched=False, reason="policy_disabled") + + for rule in self.rules: + if not isinstance(rule, dict): + continue + if not bool(rule.get("enabled", True)): + continue + + if not self._rule_matches(rule, packet, context): + continue + + action = self._resolve_action(rule) + rule_id = rule.get("id") + rule_name = rule.get("name") or "unnamed" + reason = f"Policy rule matched: id={rule_id}, name={rule_name}, action={action}" + return PolicyDecision(action=action, matched=True, rule_id=rule_id, reason=reason) + + return PolicyDecision(action=self.default_action, matched=False, reason="default_action") + + def _resolve_action(self, rule: dict) -> str: + then_block = rule.get("then", {}) + action = None + + if isinstance(then_block, dict): + action = then_block.get("action") + elif isinstance(then_block, str): + action = then_block + + if not action: + action = rule.get("action") + + action = str(action or "allow") + if action not in SUPPORTED_ACTIONS: + logger.warning("Unsupported policy action '%s', coercing to 'allow'", action) + return "allow" + return action + + def _rule_matches(self, rule: dict, packet, context: dict) -> bool: + cond = rule.get("if", {}) + + # Support implicit single-condition form. + if isinstance(cond, dict) and "field" in cond: + return self._condition_matches(cond, packet, context) + + if not isinstance(cond, dict): + return False + + all_conds = cond.get("all") + any_conds = cond.get("any") + + if isinstance(all_conds, list): + return all(self._condition_matches(c, packet, context) for c in all_conds) + + if isinstance(any_conds, list): + return any(self._condition_matches(c, packet, context) for c in any_conds) + + return False + + def _condition_matches(self, condition: dict, packet, context: dict) -> bool: + if not isinstance(condition, dict): + return False + + field = condition.get("field") + op = condition.get("op", "equals") + try: + expected = self._resolve_value(condition.get("value")) + + actual = self._get_field_value(field, packet, context) + if field == "path_hashes": + actual = self._normalize_path_hash_values(actual) + expected = self._normalize_path_hash_values(expected) + if field == "channel_hash": + actual = self._normalize_channel_hash_values(actual) + expected = self._normalize_channel_hash_values(expected) + result = self._compare(actual, op, expected) + logger.debug( + "Condition eval: field=%s op=%s expected=%r actual=%r -> %s", + field, + op, + expected, + actual, + "MATCH" if result else "no match", + ) + return result + except ValueError as exc: + logger.debug("Condition eval: field=%s raised ValueError: %s -> no match", field, exc) + return False + + def _resolve_value(self, value: Any) -> Any: + if isinstance(value, str) and value.startswith("@"): + # Object reference format: @group.name + ref = value[1:] + parts = ref.split(".", 1) + if len(parts) == 2: + group, key = parts + group_obj = self.objects.get(group, {}) + if isinstance(group_obj, dict): + return group_obj.get(key) + return value + + def _get_field_value(self, field: Any, packet, context: dict) -> Any: + if not isinstance(field, str): + return None + + # Existing packet/context fields only. + if field in context: + return context.get(field) + + if field == "payload_hex": + payload = getattr(packet, "payload", None) or b"" + return bytes(payload).hex() + + if field == "channel_hash": + decrypted = getattr(packet, "decrypted", None) + if isinstance(decrypted, dict): + group_text = decrypted.get("group_text_data", {}) + if isinstance(group_text, dict): + candidate = group_text.get("channel_hash") + if candidate is not None: + return candidate + + try: + payload_type = ( + packet.get_payload_type() if hasattr(packet, "get_payload_type") else None + ) + except Exception: + payload_type = None + if payload_type in (PAYLOAD_TYPE_GRP_TXT, PAYLOAD_TYPE_GRP_DATA): + payload = ( + packet.get_payload() + if hasattr(packet, "get_payload") + else getattr(packet, "payload", None) + ) + if payload and len(payload) >= 1: + return payload[0] + + if field == "channel_message_body": + channel_info = self._get_channel_decrypt_info(packet) + return channel_info.get("message_body") + + if field == "channel_decryptable": + channel_info = self._get_channel_decrypt_info(packet) + return bool(channel_info.get("decryptable", False)) + + if field == "path_hashes": + if hasattr(packet, "get_path_hashes_hex"): + return packet.get_path_hashes_hex() + return [] + + if field == "transport_code_0": + if hasattr(packet, "transport_codes") and packet.transport_codes: + return packet.transport_codes[0] + return None + + if field == "transport_code_1": + if hasattr(packet, "transport_codes") and len(packet.transport_codes) > 1: + return packet.transport_codes[1] + return None + + return None + + def _extract_channel_message_body(self, packet) -> Optional[str]: + channel_info = self._compute_channel_decrypt_info(packet) + return channel_info.get("message_body") + + def _get_channel_decrypt_info(self, packet) -> dict[str, Any]: + packet_key = id(packet) + cached = self._channel_decrypt_cache.get(packet_key) + if isinstance(cached, dict): + return cached + + computed = self._compute_channel_decrypt_info(packet) + self._channel_decrypt_cache[packet_key] = computed + return computed + + def _compute_channel_decrypt_info(self, packet) -> dict[str, Any]: + decrypted = getattr(packet, "decrypted", None) + if isinstance(decrypted, dict): + group_text = decrypted.get("group_text_data", {}) + if isinstance(group_text, dict): + text = group_text.get("text") + if isinstance(text, str): + return { + "decryptable": True, + "message_body": text, + } + + try: + payload_type = ( + packet.get_payload_type() if hasattr(packet, "get_payload_type") else None + ) + except Exception: + return { + "decryptable": False, + "message_body": None, + } + + if payload_type != PAYLOAD_TYPE_GRP_TXT: + return { + "decryptable": False, + "message_body": None, + } + + payload = ( + packet.get_payload() + if hasattr(packet, "get_payload") + else getattr(packet, "payload", None) + ) + if not payload or len(payload) < 4: + return { + "decryptable": False, + "message_body": None, + } + + channel_hash = payload[0] + cipher_mac = bytes(payload[1:3]) + ciphertext = bytes(payload[3:]) + + secrets_tried = 0 + for secret in self._iter_policy_channel_secrets(): + secrets_tried += 1 + derived = self._derive_channel_hash(secret) + secret_preview = secret[:8] + "..." if len(secret) > 8 else secret + if derived != channel_hash: + logger.debug( + "Channel decrypt: secret %s derived hash 0x%02X != packet hash 0x%02X, skipping", + secret_preview, + derived, + channel_hash, + ) + continue + + logger.debug( + "Channel decrypt: secret %s hash matches 0x%02X, attempting MAC+decrypt", + secret_preview, + channel_hash, + ) + plaintext = self._decrypt_channel_message(secret, cipher_mac, ciphertext) + if plaintext is None: + logger.debug( + "Channel decrypt: secret %s MAC/decrypt failed", + secret_preview, + ) + continue + + parsed = self._parse_channel_plaintext(plaintext) + if not isinstance(parsed, dict): + logger.debug( + "Channel decrypt: secret %s parse failed", + secret_preview, + ) + continue + + content = parsed.get("content") + if not isinstance(content, str): + continue + + _, message_body = self._extract_sender_from_message(content) + logger.debug( + "Channel decrypt: SUCCESS with secret %s, message_body=%r", + secret_preview, + message_body[:40] if message_body else "", + ) + return { + "decryptable": True, + "message_body": message_body.rstrip("\x00").rstrip(), + } + + if secrets_tried == 0: + logger.debug( + "Channel decrypt: no policy channel secrets configured " + "(objects.channels / objects.channel_hash_groups and inline rule secrets are empty/missing); " + "decryptable=False", + ) + else: + logger.debug( + "Channel decrypt: no matching secret found (tried %d), decryptable=False", + secrets_tried, + ) + return { + "decryptable": False, + "message_body": None, + } + + def _iter_policy_channel_secrets(self): + channels = self.objects.get("channels", {}) + if isinstance(channels, dict): + channel_items = channels.values() + elif isinstance(channels, (list, tuple)): + channel_items = channels + else: + logger.debug( + "Channel decrypt: objects.channels has unsupported type %s", + type(channels).__name__, + ) + return + + for channel_cfg in channel_items: + if isinstance(channel_cfg, str): + yield channel_cfg + continue + + if isinstance(channel_cfg, dict): + # Accept common schema variations used across policy/companion exports. + secret = ( + channel_cfg.get("secret") + or channel_cfg.get("key") + or channel_cfg.get("psk") + or channel_cfg.get("channel_secret") + ) + if secret: + yield str(secret) + else: + logger.debug( + "Channel decrypt: channel entry missing secret/key/psk/channel_secret keys", + ) + continue + + logger.debug( + "Channel decrypt: skipping unsupported channel entry type %s", + type(channel_cfg).__name__, + ) + + # Also accept full channel secrets provided via policy object groups. + channel_hash_groups = self.objects.get("channel_hash_groups", {}) + if isinstance(channel_hash_groups, dict): + for group_values in channel_hash_groups.values(): + values = ( + group_values if isinstance(group_values, (list, tuple, set)) else [group_values] + ) + for candidate in values: + secret = self._extract_channel_secret_literal(candidate) + if secret: + yield secret + elif channel_hash_groups not in ({}, None): + logger.debug( + "Channel decrypt: objects.channel_hash_groups has unsupported type %s", + type(channel_hash_groups).__name__, + ) + + for inline_secret in self._inline_channel_secrets: + yield inline_secret + + @staticmethod + def _secret_bytes_for_hash(channel_secret: str) -> bytes: + try: + secret_bytes = bytes.fromhex(channel_secret) + except ValueError: + secret_bytes = channel_secret.encode("utf-8") + if len(secret_bytes) >= 32 and secret_bytes[16:32] == b"\x00" * 16: + return secret_bytes[:16] + if len(secret_bytes) > 32: + return secret_bytes[:32] + return secret_bytes + + def _derive_channel_hash(self, channel_secret: str) -> int: + secret_bytes = self._secret_bytes_for_hash(channel_secret) + return hashlib.sha256(secret_bytes).digest()[0] + + @staticmethod + def _decrypt_channel_message( + channel_secret: str, mac: bytes, ciphertext: bytes + ) -> Optional[bytes]: + try: + try: + secret_bytes = bytes.fromhex(channel_secret) + except ValueError: + secret_bytes = channel_secret.encode("utf-8") + + if len(secret_bytes) < 32: + secret_bytes = secret_bytes + b"\x00" * (32 - len(secret_bytes)) + elif len(secret_bytes) > 32: + secret_bytes = secret_bytes[:32] + + expected_mac = CryptoUtils._hmac_sha256(secret_bytes, ciphertext)[:2] + if mac != expected_mac: + return None + + return CryptoUtils._aes_decrypt(secret_bytes[:16], ciphertext) + except Exception: + return None + + @staticmethod + def _parse_channel_plaintext(plaintext: bytes) -> Optional[dict]: + if len(plaintext) < 5: + return None + + try: + timestamp = int.from_bytes(plaintext[:4], "little") + flags = plaintext[4] + raw = plaintext[5:].decode("utf-8", errors="replace") + message_content = raw.rstrip("\x00") + + message_type = "unknown" + if flags == 0x00: + message_type = "plain_text" + elif flags == 0x01: + message_type = "cli_command" + elif flags == 0x02: + message_type = "signed_text" + if len(plaintext) >= 7: + raw = plaintext[7:].decode("utf-8", errors="replace") + message_content = raw.rstrip("\x00") + + return { + "timestamp": timestamp, + "flags": flags, + "message_type": message_type, + "content": message_content, + } + except Exception: + return None + + @staticmethod + def _extract_sender_from_message(message_content: str) -> tuple[str, str]: + if ": " in message_content: + parts = message_content.split(": ", 1) + if len(parts) == 2: + return parts[0], parts[1] + return "Unknown", message_content + + @staticmethod + def _extract_channel_secret_literal(value: Any) -> Optional[str]: + if isinstance(value, str): + raw = value.strip() + if not raw: + return None + normalized_hex = raw[2:] if raw.lower().startswith("0x") else raw + if len(normalized_hex) in (32, 64) and all( + ch in "0123456789abcdefABCDEF" for ch in normalized_hex + ): + return normalized_hex + return None + + @classmethod + def _collect_inline_rule_channel_secrets(cls, rules: list) -> list[str]: + secrets: list[str] = [] + + def _consume_condition(cond: Any): + if not isinstance(cond, dict): + return + if cond.get("field") != "channel_hash": + return + raw_value = cond.get("value") + values = raw_value if isinstance(raw_value, (list, tuple, set)) else [raw_value] + for candidate in values: + secret = cls._extract_channel_secret_literal(candidate) + if secret: + secrets.append(secret) + + for rule in rules: + if not isinstance(rule, dict): + continue + cond = rule.get("if", {}) + if isinstance(cond, dict) and "field" in cond: + _consume_condition(cond) + continue + if not isinstance(cond, dict): + continue + for key in ("all", "any"): + branch = cond.get(key) + if isinstance(branch, list): + for item in branch: + _consume_condition(item) + + return list(dict.fromkeys(secrets)) + + @staticmethod + def _normalize_path_hash_value(value: Any) -> Optional[str]: + if value is None: + return None + + if isinstance(value, int): + parsed = value + if parsed < 0: + return None + if parsed <= 0xFF: + width = 2 + elif parsed <= 0xFFFF: + width = 4 + elif parsed <= 0xFFFFFF: + width = 6 + else: + raise ValueError("path hash exceeds 3 bytes") + return f"{parsed:0{width}X}" + else: + raw = str(value).strip() + if not raw: + return None + if raw.lower().startswith("0x"): + raw = raw[2:] + if not raw: + return None + if len(raw) % 2 != 0: + raise ValueError("path hash hex length must be even") + if len(raw) not in (2, 4, 6): + raise ValueError("path hash must be 1, 2, or 3 bytes") + if not all(ch in "0123456789abcdefABCDEF" for ch in raw): + raise ValueError("path hash must be hex") + return raw.upper() + + @classmethod + def _normalize_path_hash_values(cls, value: Any) -> Any: + if isinstance(value, (list, tuple, set)): + normalized = [] + for item in value: + normalized_item = cls._normalize_path_hash_value(item) + if normalized_item is not None: + normalized.append(normalized_item) + lengths = {len(item) for item in normalized} + if len(lengths) > 1: + raise ValueError("path hashes cannot mix byte lengths") + return normalized + + return cls._normalize_path_hash_value(value) + + @staticmethod + def _normalize_channel_hash_value(value: Any) -> Optional[str]: + if value is None: + return None + + if isinstance(value, int): + parsed = value + else: + raw = str(value).strip() + if not raw: + return None + normalized_hex = raw[2:] if raw.lower().startswith("0x") else raw + if len(normalized_hex) in (32, 64) and all( + ch in "0123456789abcdefABCDEF" for ch in normalized_hex + ): + secret_bytes = PolicyEngine._secret_bytes_for_hash(normalized_hex) + parsed = hashlib.sha256(secret_bytes).digest()[0] + return f"0x{parsed:02X}" + if raw.lower().startswith("0x"): + parsed = int(raw, 16) + elif raw.isdigit(): + parsed = int(raw, 10) + else: + parsed = int(raw, 16) + + if parsed < 0: + raise ValueError("channel hash must be non-negative") + if parsed > 0xFF: + raise ValueError("channel hash must be one byte (0x00-0xFF)") + return f"0x{parsed:02X}" + + @classmethod + def _normalize_channel_hash_values(cls, value: Any) -> Any: + if isinstance(value, (list, tuple, set)): + normalized = [] + for item in value: + normalized_item = cls._normalize_channel_hash_value(item) + if normalized_item is not None: + normalized.append(normalized_item) + return normalized + + return cls._normalize_channel_hash_value(value) + + @staticmethod + def _compare(actual: Any, op: Any, expected: Any) -> bool: + op_name = str(op or "equals").lower() + + try: + if op_name in ("equals", "eq", "=="): + return actual == expected + if op_name in ("not_equals", "ne", "!="): + return actual != expected + if op_name in ("greater_than", "gt", ">"): + return actual is not None and expected is not None and actual > expected + if op_name in ("greater_or_equal", "gte", ">="): + return actual is not None and expected is not None and actual >= expected + if op_name in ("less_than", "lt", "<"): + return actual is not None and expected is not None and actual < expected + if op_name in ("less_or_equal", "lte", "<="): + return actual is not None and expected is not None and actual <= expected + if op_name == "contains": + if isinstance(actual, (list, tuple, set)): + return expected in actual + if isinstance(actual, str) and expected is not None: + return str(expected) in actual + return False + if op_name in ("in", "is_in"): + if isinstance(expected, (list, tuple, set)): + return actual in expected + if isinstance(expected, str) and actual is not None: + return str(actual) in expected + return False + if op_name in ("intersects", "overlaps"): + if isinstance(actual, (list, tuple, set)) and isinstance( + expected, (list, tuple, set) + ): + return len(set(actual).intersection(set(expected))) > 0 + return False + if op_name == "starts_with": + return ( + isinstance(actual, str) + and isinstance(expected, str) + and actual.startswith(expected) + ) + if op_name == "ends_with": + return ( + isinstance(actual, str) + and isinstance(expected, str) + and actual.endswith(expected) + ) + except Exception: + return False + + return False diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index 200ae65..ff998bc 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -1,12 +1,14 @@ import json import logging import os +import re import secrets import time from datetime import datetime, timezone from typing import Callable, Optional import cherrypy +import yaml from pymc_core.protocol import CryptoUtils from repeater import __version__ @@ -16,6 +18,7 @@ from repeater.companion.identity_resolve import ( heal_companion_empty_names, ) from repeater.config import resolve_storage_dir +from repeater.policy_engine import PolicyEngine from repeater.service_utils import get_buildroot_image_info from .auth.middleware import require_auth @@ -26,6 +29,14 @@ from .update_endpoints import UpdateAPIEndpoints logger = logging.getLogger("HTTPServer") +POLICY_GROUP_KINDS = { + "channel_hash": "channel_hashes", + "channel_hashes": "channel_hashes", + "channels": "channel_hashes", + "pubkey": "pubkeys", + "pubkeys": "pubkeys", +} + # ============================================================================ # API ENDPOINT DOCUMENTATION @@ -45,6 +56,7 @@ logger = logging.getLogger("HTTPServer") # GET /api/gps - Get local GPS diagnostics and parsed NMEA attributes # GET /api/gps_stream - GPS diagnostics SSE stream # GET /api/logs - Get system logs +# GET /api/logs_stream - Stream live system logs over SSE # GET /api/hardware_stats - Get hardware statistics # GET /api/hardware_processes - Get process information # GET /api/validate_config - Validate config.yaml syntax and required settings @@ -319,6 +331,623 @@ class APIEndpoints: } return has_default_name or has_default_password or radio_not_configured, reasons + def _default_policy_document(self) -> dict: + return { + "policy_engine": { + "enabled": False, + "default_action": "allow", + "rules": [], + "objects": {}, + }, + "groups": { + "channel_hashes": [], + "pubkeys": [], + }, + } + + @staticmethod + def _slugify_policy_id(value: str, fallback: str) -> str: + text = str(value or "").strip().lower() + text = re.sub(r"[^a-z0-9]+", "_", text).strip("_") + return text or fallback + + def _normalize_policy_group_kind(self, raw_kind) -> Optional[str]: + if raw_kind is None: + return None + key = str(raw_kind).strip().lower() + return POLICY_GROUP_KINDS.get(key) + + def _normalize_pubkey_value(self, value) -> str: + if value is None: + raise ValueError("pubkey value is required") + if isinstance(value, bytes): + raw = value.hex() + else: + raw = str(value).strip().lower() + if raw.startswith("0x"): + raw = raw[2:] + raw = raw.replace(" ", "") + if not raw: + raise ValueError("pubkey value is required") + if not re.fullmatch(r"[0-9a-f]+", raw): + raise ValueError("pubkey must be hex") + if len(raw) % 2 != 0: + raise ValueError("pubkey hex length must be even") + return f"0x{raw}" + + def _normalize_channel_hash_value(self, value) -> str: + if value is None: + raise ValueError("channel hash value is required") + + if isinstance(value, int): + parsed = value + else: + raw = str(value).strip() + if not raw: + raise ValueError("channel hash value is required") + normalized_hex = raw[2:] if raw.lower().startswith("0x") else raw + if len(normalized_hex) in (32, 64) and re.fullmatch(r"[0-9a-fA-F]+", normalized_hex): + return f"0x{normalized_hex.upper()}" + if raw.lower().startswith("0x"): + parsed = int(raw, 16) + elif re.fullmatch(r"[0-9]+", raw): + parsed = int(raw, 10) + else: + parsed = int(raw, 16) + + if parsed < 0: + raise ValueError("channel hash must be non-negative") + if parsed > 0xFF: + raise ValueError("channel hash must be one byte (0x00-0xFF)") + return f"0x{parsed:02X}" + + def _normalize_policy_entry_value(self, kind: str, value) -> str: + if kind == "pubkeys": + return self._normalize_pubkey_value(value) + if kind == "channel_hashes": + return self._normalize_channel_hash_value(value) + raise ValueError(f"Unsupported group kind: {kind}") + + def _normalize_policy_groups(self, groups_cfg: dict) -> dict: + normalized = {"channel_hashes": [], "pubkeys": []} + + if not isinstance(groups_cfg, dict): + return normalized + + for kind in ("channel_hashes", "pubkeys"): + source_groups = groups_cfg.get(kind) + if not isinstance(source_groups, list): + continue + + seen_group_ids = set() + for idx, group in enumerate(source_groups): + if not isinstance(group, dict): + continue + group_id = self._slugify_policy_id( + group.get("id") or group.get("name") or group.get("friendly_name"), + f"{kind}_{idx + 1}", + ) + if group_id in seen_group_ids: + continue + seen_group_ids.add(group_id) + + friendly_name = str(group.get("friendly_name") or group.get("name") or group_id) + description = str(group.get("description") or "") + entries = [] + seen_entry_ids = set() + + for ent_idx, entry in enumerate(group.get("entries") or []): + if not isinstance(entry, dict): + continue + try: + entry_value = self._normalize_policy_entry_value(kind, entry.get("value")) + except Exception: + continue + + entry_id = self._slugify_policy_id( + entry.get("id") + or entry.get("name") + or entry.get("friendly_name") + or entry_value, + f"entry_{ent_idx + 1}", + ) + if entry_id in seen_entry_ids: + continue + seen_entry_ids.add(entry_id) + + entry_friendly_name = str( + entry.get("friendly_name") or entry.get("name") or entry_id + ) + entries.append( + { + "id": entry_id, + "friendly_name": entry_friendly_name, + "value": entry_value, + } + ) + + normalized[kind].append( + { + "id": group_id, + "friendly_name": friendly_name, + "description": description, + "entries": entries, + } + ) + + return normalized + + def _policy_objects_from_groups(self, groups_cfg: dict) -> dict: + channel_hash_groups = {} + pubkey_groups = {} + + for group in groups_cfg.get("channel_hashes", []): + channel_hash_groups[group["id"]] = [ + entry["value"] for entry in group.get("entries", []) + ] + + for group in groups_cfg.get("pubkeys", []): + pubkey_groups[group["id"]] = [entry["value"] for entry in group.get("entries", [])] + + return { + "channel_hash_groups": channel_hash_groups, + "pubkey_groups": pubkey_groups, + } + + def _write_policy_document(self, doc: dict) -> None: + policy_path = self._get_policy_file_path() + os.makedirs(os.path.dirname(policy_path), exist_ok=True) + with open(policy_path, "w", encoding="utf-8") as f: + yaml.safe_dump( + doc, + f, + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + width=1000000, + ) + + def _sync_policy_engine_objects_from_groups(self, doc: dict) -> dict: + policy_engine_cfg = self._normalize_policy_engine(doc.get("policy_engine", {})) + groups_cfg = self._normalize_policy_groups(doc.get("groups", {})) + + objects = policy_engine_cfg.get("objects", {}) + if not isinstance(objects, dict): + objects = {} + objects.update(self._policy_objects_from_groups(groups_cfg)) + + policy_engine_cfg["objects"] = objects + doc["policy_engine"] = policy_engine_cfg + doc["groups"] = groups_cfg + return doc + + def _get_policy_file_path(self) -> str: + policy_cfg = self.config.get("policy", {}) if isinstance(self.config, dict) else {} + policy_file = policy_cfg.get("policy_file", "policy.yaml") + + config_dir = os.path.dirname(os.path.abspath(self._config_path)) + if os.path.isabs(str(policy_file)): + return str(policy_file) + return os.path.abspath(os.path.join(config_dir, str(policy_file))) + + def _load_policy_document(self) -> tuple[dict, bool]: + path = self._get_policy_file_path() + if not os.path.exists(path): + return self._default_policy_document(), False + + try: + with open(path, "r", encoding="utf-8") as f: + data = yaml.safe_load(f) or {} + if not isinstance(data, dict): + return self._default_policy_document(), False + if "policy_engine" not in data: + return {"policy_engine": data}, True + if not isinstance(data.get("policy_engine"), dict): + return self._default_policy_document(), False + return data, True + except Exception as e: + logger.error(f"Failed to load policy file {path}: {e}") + return self._default_policy_document(), False + + @staticmethod + def _normalize_policy_engine(engine_cfg: dict) -> dict: + normalized = { + "enabled": bool(engine_cfg.get("enabled", False)), + "default_action": str(engine_cfg.get("default_action", "allow")), + "rules": engine_cfg.get("rules") if isinstance(engine_cfg.get("rules"), list) else [], + "objects": ( + engine_cfg.get("objects") if isinstance(engine_cfg.get("objects"), dict) else {} + ), + } + return normalized + + def _apply_policy_runtime(self, policy_engine_cfg: dict) -> None: + self.config["policy_engine"] = policy_engine_cfg + self.config["policy_file_path"] = self._get_policy_file_path() + + if not self.daemon_instance: + return + repeater_handler = getattr(self.daemon_instance, "repeater_handler", None) + if not repeater_handler: + return + try: + repeater_handler.policy_engine = PolicyEngine.from_runtime_config(self.config) + except Exception as e: + logger.warning(f"Failed to apply runtime policy engine update: {e}") + + @cherrypy.expose + @cherrypy.tools.json_out() + @cherrypy.tools.json_in() + def policy(self, **kwargs): + """GET/POST policy.yaml next to config.yaml. + + GET: returns policy document and file metadata. + POST: writes policy document and applies runtime policy engine refresh. + """ + self._set_cors_headers() + if cherrypy.request.method == "OPTIONS": + return "" + + policy_path = self._get_policy_file_path() + + if cherrypy.request.method == "GET": + doc, exists = self._load_policy_document() + doc = self._sync_policy_engine_objects_from_groups(doc) + normalized = self._normalize_policy_engine(doc.get("policy_engine", {})) + return self._success( + { + "policy_file": policy_path, + "exists": exists, + "policy_engine": normalized, + "groups": doc.get("groups", self._default_policy_document().get("groups", {})), + } + ) + + if cherrypy.request.method != "POST": + return self._error("Method not supported") + + try: + self._require_post() + body = cherrypy.request.json or {} + + if not isinstance(body, dict): + return self._error("Invalid payload: expected JSON object") + + if isinstance(body.get("policy_engine"), dict): + policy_engine_cfg = body.get("policy_engine") + else: + # Allow posting the policy_engine object directly. + policy_engine_cfg = body + + existing_doc, _ = self._load_policy_document() + groups_from_body = body.get("groups") + if groups_from_body is None: + normalized_groups = self._normalize_policy_groups(existing_doc.get("groups", {})) + else: + normalized_groups = self._normalize_policy_groups(groups_from_body) + + normalized = self._normalize_policy_engine(policy_engine_cfg) + if "objects" not in policy_engine_cfg and isinstance( + existing_doc.get("policy_engine", {}).get("objects"), dict + ): + normalized["objects"] = dict( + existing_doc.get("policy_engine", {}).get("objects", {}) + ) + + doc_to_write = { + "policy_engine": normalized, + "groups": normalized_groups, + } + doc_to_write = self._sync_policy_engine_objects_from_groups(doc_to_write) + self._write_policy_document(doc_to_write) + + # Validate via PolicyEngine construction and apply live. + PolicyEngine(doc_to_write.get("policy_engine", {})) + self._apply_policy_runtime(doc_to_write.get("policy_engine", {})) + + logger.info("Policy updated and saved to %s", policy_path) + return self._success( + { + "policy_file": policy_path, + "policy_engine": doc_to_write.get("policy_engine", {}), + "groups": doc_to_write.get("groups", {}), + }, + message="Policy updated and applied", + restart_required=False, + ) + except cherrypy.HTTPError: + raise + except Exception as e: + logger.error(f"Error updating policy: {e}", exc_info=True) + return self._error(str(e)) + + @cherrypy.expose + @cherrypy.tools.json_out() + @cherrypy.tools.json_in() + def policy_validate(self): + """Validate policy payload without saving it.""" + self._set_cors_headers() + if cherrypy.request.method == "OPTIONS": + return "" + + try: + self._require_post() + body = cherrypy.request.json or {} + if not isinstance(body, dict): + return self._error("Invalid payload: expected JSON object") + + if isinstance(body.get("policy_engine"), dict): + policy_engine_cfg = body.get("policy_engine") + else: + policy_engine_cfg = body + + normalized = self._normalize_policy_engine(policy_engine_cfg) + engine = PolicyEngine(normalized) + + return self._success( + { + "valid": True, + "normalized": normalized, + "effective": { + "enabled": engine.enabled, + "default_action": engine.default_action, + "rule_count": len(engine.rules), + }, + } + ) + except cherrypy.HTTPError: + raise + except Exception as e: + logger.error(f"Policy validation failed: {e}") + return self._success({"valid": False, "error": str(e)}) + + @cherrypy.expose + @cherrypy.tools.json_out() + @cherrypy.tools.json_in(force=False) + def policy_groups(self, **kwargs): + """Manage policy groups for channel hashes and pubkeys.""" + self._set_cors_headers() + if cherrypy.request.method == "OPTIONS": + return "" + + if cherrypy.request.method == "GET": + kind = self._normalize_policy_group_kind(cherrypy.request.params.get("kind")) + if cherrypy.request.params.get("kind") and not kind: + return self._error("Invalid kind. Use 'channel_hashes' or 'pubkeys'") + + doc, exists = self._load_policy_document() + doc = self._sync_policy_engine_objects_from_groups(doc) + groups = doc.get("groups", self._default_policy_document().get("groups", {})) + + data = { + "policy_file": self._get_policy_file_path(), + "exists": exists, + "kind": kind, + "groups": groups[kind] if kind else groups, + } + return self._success(data) + + if cherrypy.request.method not in ("POST", "DELETE"): + return self._error("Method not supported") + + try: + body = cherrypy.request.json or {} + if not isinstance(body, dict): + body = {} + + request_params = getattr(cherrypy.request, "params", {}) or {} + kind = self._normalize_policy_group_kind(body.get("kind") or request_params.get("kind")) + if not kind: + return self._error("Invalid kind. Use 'channel_hashes' or 'pubkeys'") + + group_id = self._slugify_policy_id( + body.get("group_id") + or request_params.get("group_id") + or body.get("id") + or body.get("name") + or body.get("friendly_name"), + "group", + ) + + doc, _ = self._load_policy_document() + doc = self._sync_policy_engine_objects_from_groups(doc) + groups = doc.get("groups", self._default_policy_document().get("groups", {})) + group_list = groups.get(kind, []) + + existing_group = next((g for g in group_list if g.get("id") == group_id), None) + + if cherrypy.request.method == "POST": + if existing_group: + return self._error(f"Group already exists: {group_id}") + + friendly_name = str(body.get("friendly_name") or body.get("name") or group_id) + description = str(body.get("description") or "") + new_group = { + "id": group_id, + "friendly_name": friendly_name, + "description": description, + "entries": [], + } + group_list.append(new_group) + + groups[kind] = group_list + doc["groups"] = self._normalize_policy_groups(groups) + doc = self._sync_policy_engine_objects_from_groups(doc) + self._write_policy_document(doc) + self._apply_policy_runtime(doc.get("policy_engine", {})) + + return self._success( + { + "kind": kind, + "group": new_group, + "groups": doc.get("groups", {}), + }, + message="Policy group created", + restart_required=False, + ) + + if not existing_group: + return self._error(f"Group not found: {group_id}") + + group_list = [g for g in group_list if g.get("id") != group_id] + groups[kind] = group_list + doc["groups"] = self._normalize_policy_groups(groups) + doc = self._sync_policy_engine_objects_from_groups(doc) + self._write_policy_document(doc) + self._apply_policy_runtime(doc.get("policy_engine", {})) + + return self._success( + { + "kind": kind, + "group_id": group_id, + "groups": doc.get("groups", {}), + }, + message="Policy group deleted", + restart_required=False, + ) + except Exception as e: + logger.error(f"Error managing policy groups: {e}", exc_info=True) + return self._error(str(e)) + + @cherrypy.expose + @cherrypy.tools.json_out() + @cherrypy.tools.json_in(force=False) + def policy_group_entries(self, **kwargs): + """Manage entries inside policy groups.""" + self._set_cors_headers() + if cherrypy.request.method == "OPTIONS": + return "" + + if cherrypy.request.method == "GET": + kind = self._normalize_policy_group_kind(cherrypy.request.params.get("kind")) + if not kind: + return self._error("Invalid kind. Use 'channel_hashes' or 'pubkeys'") + + group_id = self._slugify_policy_id(cherrypy.request.params.get("group_id"), "group") + doc, _ = self._load_policy_document() + doc = self._sync_policy_engine_objects_from_groups(doc) + groups = doc.get("groups", self._default_policy_document().get("groups", {})) + group = next((g for g in groups.get(kind, []) if g.get("id") == group_id), None) + if not group: + return self._error(f"Group not found: {group_id}") + + return self._success( + { + "kind": kind, + "group": group, + "entries": group.get("entries", []), + } + ) + + if cherrypy.request.method not in ("POST", "DELETE"): + return self._error("Method not supported") + + try: + body = cherrypy.request.json or {} + if not isinstance(body, dict): + body = {} + + request_params = getattr(cherrypy.request, "params", {}) or {} + kind = self._normalize_policy_group_kind(body.get("kind") or request_params.get("kind")) + if not kind: + return self._error("Invalid kind. Use 'channel_hashes' or 'pubkeys'") + + group_id = self._slugify_policy_id( + body.get("group_id") or request_params.get("group_id") or body.get("group"), + "group", + ) + + doc, _ = self._load_policy_document() + doc = self._sync_policy_engine_objects_from_groups(doc) + groups = doc.get("groups", self._default_policy_document().get("groups", {})) + group = next((g for g in groups.get(kind, []) if g.get("id") == group_id), None) + if not group: + return self._error(f"Group not found: {group_id}") + + entries = list(group.get("entries", [])) + + if cherrypy.request.method == "POST": + entry_value = self._normalize_policy_entry_value(kind, body.get("value")) + if any(entry.get("value") == entry_value for entry in entries): + return self._error(f"Entry already exists in group: {entry_value}") + + entry_id = self._slugify_policy_id( + body.get("entry_id") + or body.get("id") + or body.get("name") + or body.get("friendly_name") + or entry_value, + "entry", + ) + if any(entry.get("id") == entry_id for entry in entries): + return self._error(f"Entry id already exists in group: {entry_id}") + + new_entry = { + "id": entry_id, + "friendly_name": str(body.get("friendly_name") or body.get("name") or entry_id), + "value": entry_value, + } + entries.append(new_entry) + group["entries"] = entries + + doc["groups"] = self._normalize_policy_groups(groups) + doc = self._sync_policy_engine_objects_from_groups(doc) + self._write_policy_document(doc) + self._apply_policy_runtime(doc.get("policy_engine", {})) + + return self._success( + { + "kind": kind, + "group_id": group_id, + "entry": new_entry, + "group": group, + }, + message="Policy group entry added", + restart_required=False, + ) + + entry_id = body.get("entry_id") or request_params.get("entry_id") or body.get("id") + value = body.get("value") or request_params.get("value") + if not entry_id and value is None: + return self._error("Provide entry_id or value") + + if value is not None: + value = self._normalize_policy_entry_value(kind, value) + + filtered_entries = [] + removed = None + for entry in entries: + if entry_id and entry.get("id") == entry_id: + removed = entry + continue + if value is not None and entry.get("value") == value: + removed = entry + continue + filtered_entries.append(entry) + + if removed is None: + return self._error("Entry not found") + + group["entries"] = filtered_entries + doc["groups"] = self._normalize_policy_groups(groups) + doc = self._sync_policy_engine_objects_from_groups(doc) + self._write_policy_document(doc) + self._apply_policy_runtime(doc.get("policy_engine", {})) + + return self._success( + { + "kind": kind, + "group_id": group_id, + "removed": removed, + "group": group, + }, + message="Policy group entry removed", + restart_required=False, + ) + except Exception as e: + logger.error(f"Error managing policy group entries: {e}", exc_info=True) + return self._error(str(e)) + # ============================================================================ # SETUP WIZARD ENDPOINTS # ============================================================================ @@ -329,8 +958,6 @@ class APIEndpoints: """Check if the repeater needs initial setup configuration""" try: # Prefer the on-disk config so this reflects current persisted state. - import yaml - config = self.config config_path = getattr(self, "_config_path", None) try: @@ -488,8 +1115,6 @@ class APIEndpoints: self._require_post() data = cherrypy.request.json - import yaml - # Setup wizard is first-run only. After setup, use /auth/change_password # and /api/update_radio_config for subsequent changes. try: @@ -1560,8 +2185,6 @@ class APIEndpoints: raise cherrypy.HTTPError(405, "Method not allowed. This endpoint requires GET.") try: - import yaml - errors = [] warnings = [] @@ -1866,16 +2489,22 @@ class APIEndpoints: from .http_server import _log_buffer try: - logs = list(_log_buffer.logs) + if hasattr(_log_buffer, "snapshot"): + logs = _log_buffer.snapshot() + else: + logs = list(getattr(_log_buffer, "logs", [])) return { "logs": ( logs if logs else [ { + "id": 0, "message": "No logs available", "timestamp": datetime.now().isoformat(), "level": "INFO", + "logger": "HTTPServer", + "raw_message": "No logs available", } ] ) @@ -1884,6 +2513,89 @@ class APIEndpoints: logger.error(f"Error fetching logs: {e}") return {"error": str(e), "logs": []} + @cherrypy.expose + def logs_stream(self, since_id: Optional[str] = None): + from .http_server import _log_buffer + + cherrypy.response.headers["Content-Type"] = "text/event-stream" + cherrypy.response.headers["Cache-Control"] = "no-cache" + cherrypy.response.headers["Connection"] = "keep-alive" + cherrypy.response.headers["X-Accel-Buffering"] = "no" + + if since_id is None: + since_id = cherrypy.request.headers.get("Last-Event-ID") + + try: + cursor = int(since_id) if since_id is not None else None + except (TypeError, ValueError): + cursor = None + + def encode_event(payload, event_name: Optional[str] = None, event_id: Optional[int] = None): + lines = [] + if event_name: + lines.append(f"event: {event_name}") + if event_id is not None: + lines.append(f"id: {event_id}") + lines.append(f"data: {json.dumps(payload, default=str)}") + return "\n".join(lines) + "\n\n" + + def generate(): + subscriber = _log_buffer.subscribe() + last_sent_id = cursor + current_snapshot = _log_buffer.snapshot() + + try: + yield encode_event( + { + "type": "connected", + "message": "Connected to logs stream", + "latest_id": current_snapshot[-1]["id"] if current_snapshot else 0, + }, + event_name="connected", + ) + + backlog = _log_buffer.snapshot(since_id=cursor) + for entry in backlog: + last_sent_id = entry.get("id", last_sent_id) + yield encode_event( + {"type": "log", "entry": entry}, event_name="log", event_id=entry.get("id") + ) + + while True: + try: + entry = subscriber.get(timeout=15.0) + except Exception: + yield encode_event( + {"type": "keepalive"}, event_name="keepalive", event_id=last_sent_id + ) + continue + + entry_id = entry.get("id") + if ( + last_sent_id is not None + and entry_id is not None + and entry_id <= last_sent_id + ): + continue + + last_sent_id = entry_id + yield encode_event( + {"type": "log", "entry": entry}, event_name="log", event_id=entry_id + ) + except GeneratorExit: + logger.debug("Logs SSE stream closed by client") + except Exception as exc: + logger.error(f"Error serving logs stream: {exc}", exc_info=True) + yield encode_event( + {"type": "error", "error": str(exc)}, event_name="error", event_id=last_sent_id + ) + finally: + _log_buffer.unsubscribe(subscriber) + + return generate() + + logs_stream._cp_config = {"response.stream": True} + @cherrypy.expose @cherrypy.tools.json_out() def hardware_stats(self): @@ -2312,6 +3024,7 @@ class APIEndpoints: "rx_count": "Received Packets", "tx_count": "Transmitted Packets", "drop_count": "Dropped Packets", + "policy_events": "Policy Events", "avg_rssi": "Average RSSI (dBm)", "avg_snr": "Average SNR (dB)", "avg_length": "Average Packet Length", @@ -2325,11 +3038,37 @@ class APIEndpoints: requested_metrics = [m.strip() for m in metrics.split(",")] else: requested_metrics = list(rrd_data["metrics"].keys()) + requested_metrics.append("policy_events") timestamps_ms = [ts * 1000 for ts in rrd_data["timestamps"]] series = [] for metric_key in requested_metrics: + if metric_key == "policy_events": + bucket_seconds = max(1, int(rrd_data.get("step", 60))) + policy_rows = self._get_storage().get_policy_event_counts( + start_timestamp=rrd_data["start_time"], + end_timestamp=rrd_data["end_time"], + bucket_seconds=bucket_seconds, + ) + policy_by_bucket = { + int(row.get("timestamp", 0)): int(row.get("count", 0)) + for row in policy_rows + } + chart_data = [] + for ts in rrd_data["timestamps"]: + bucket_ts = int(ts / bucket_seconds) * bucket_seconds + chart_data.append([ts * 1000, policy_by_bucket.get(bucket_ts, 0)]) + + series.append( + { + "name": metric_names["policy_events"], + "type": "policy_events", + "data": chart_data, + } + ) + continue + if metric_key in rrd_data["metrics"]: if metric_key in counter_metrics: chart_data = self._process_counter_data( diff --git a/repeater/web/http_server.py b/repeater/web/http_server.py index 0b3ab57..02d926c 100644 --- a/repeater/web/http_server.py +++ b/repeater/web/http_server.py @@ -1,7 +1,9 @@ import json import logging import os +import queue import secrets +import threading from collections import deque from datetime import datetime from pathlib import Path @@ -43,25 +45,78 @@ class LogBuffer(logging.Handler): def __init__(self, max_lines=100): super().__init__() self.logs = deque(maxlen=max_lines) + self._next_id = 1 + self._lock = threading.Lock() + self._subscribers = [] self.setFormatter(logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")) def emit(self, record): try: - msg = self.format(record) - self.logs.append( - { - "message": msg, - "timestamp": datetime.fromtimestamp(record.created).isoformat(), - "level": record.levelname, - } - ) + formatted_message = self.format(record) + entry = { + "id": self._next_log_id(), + "message": formatted_message, + "timestamp": datetime.fromtimestamp(record.created).isoformat(), + "level": record.levelname, + "logger": record.name, + "raw_message": record.getMessage(), + "module": record.module, + "pathname": record.pathname, + "line": record.lineno, + "thread": record.threadName, + "process": record.processName, + } + + if record.exc_info: + entry["exception"] = self.formatException(record.exc_info) + + with self._lock: + self.logs.append(entry) + dead_subscribers = [] + for subscriber in self._subscribers: + try: + subscriber.put_nowait(entry) + except Exception: + dead_subscribers.append(subscriber) + + if dead_subscribers: + self._subscribers = [ + subscriber + for subscriber in self._subscribers + if subscriber not in dead_subscribers + ] except Exception: self.handleError(record) + def _next_log_id(self): + with self._lock: + next_id = self._next_id + self._next_id += 1 + return next_id + + def snapshot(self, since_id=None): + with self._lock: + records = list(self.logs) + + if since_id is None: + return records + + return [record for record in records if record.get("id", 0) > since_id] + + def subscribe(self): + subscriber = queue.Queue() + with self._lock: + self._subscribers.append(subscriber) + return subscriber + + def unsubscribe(self, subscriber): + with self._lock: + self._subscribers = [item for item in self._subscribers if item is not subscriber] + # Global log buffer instance -_log_buffer = LogBuffer(max_lines=100) +_log_buffer = LogBuffer(max_lines=1000) class DocEndpoint: diff --git a/repeater/web/openapi.yaml b/repeater/web/openapi.yaml index f08d1ab..45c1577 100644 --- a/repeater/web/openapi.yaml +++ b/repeater/web/openapi.yaml @@ -494,6 +494,26 @@ paths: error: type: string + /logs_stream: + get: + tags: [System] + summary: Stream system logs + description: Server-Sent Events stream of live system log entries. + parameters: + - name: since_id + in: query + required: false + schema: + type: integer + description: Resume the stream after this log entry id. + responses: + '200': + description: SSE stream + content: + text/event-stream: + schema: + type: string + /hardware_stats: get: tags: [System] @@ -1565,6 +1585,275 @@ paths: error: type: string + /policy: + get: + tags: [Network Policy] + summary: Get policy document + description: Returns normalized policy engine configuration and grouped channel hash/pubkey entries. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + responses: + '200': + description: Policy document + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + type: object + properties: + policy_file: + type: string + exists: + type: boolean + policy_engine: + type: object + groups: + type: object + post: + tags: [Network Policy] + summary: Update policy document + description: Update policy_engine configuration while preserving or replacing named groups. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + description: Accepts either root-level policy_engine fields or a nested policy_engine object. + responses: + '200': + description: Policy updated + content: + application/json: + schema: + $ref: '#/components/schemas/SuccessResponse' + + /policy_validate: + post: + tags: [Network Policy] + summary: Validate policy payload + description: Validate a policy payload without saving it to disk. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + responses: + '200': + description: Validation result + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + type: object + properties: + valid: + type: boolean + normalized: + type: object + effective: + type: object + + /policy_groups: + get: + tags: [Network Policy] + summary: List policy groups + description: List named channel hash and pubkey groups. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + parameters: + - name: kind + in: query + required: false + schema: + type: string + enum: [channel_hashes, pubkeys] + description: Optional group kind filter. + responses: + '200': + description: Policy groups + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + type: object + post: + tags: [Network Policy] + summary: Create policy group + description: Create a named group for channel hashes or pubkeys. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [kind] + properties: + kind: + type: string + enum: [channel_hashes, pubkeys] + group_id: + type: string + friendly_name: + type: string + description: + type: string + responses: + '200': + description: Group created + content: + application/json: + schema: + $ref: '#/components/schemas/SuccessResponse' + delete: + tags: [Network Policy] + summary: Delete policy group + description: Delete a named group and all of its entries. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [kind, group_id] + properties: + kind: + type: string + enum: [channel_hashes, pubkeys] + group_id: + type: string + responses: + '200': + description: Group deleted + content: + application/json: + schema: + $ref: '#/components/schemas/SuccessResponse' + + /policy_group_entries: + get: + tags: [Network Policy] + summary: List group entries + description: Return entries for a specific named channel hash/pubkey group. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + parameters: + - name: kind + in: query + required: true + schema: + type: string + enum: [channel_hashes, pubkeys] + - name: group_id + in: query + required: true + schema: + type: string + responses: + '200': + description: Group entries + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + type: object + post: + tags: [Network Policy] + summary: Add group entry + description: Add a friendly-named entry to a policy group. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [kind, group_id, value] + properties: + kind: + type: string + enum: [channel_hashes, pubkeys] + group_id: + type: string + value: + type: string + entry_id: + type: string + friendly_name: + type: string + responses: + '200': + description: Entry added + content: + application/json: + schema: + $ref: '#/components/schemas/SuccessResponse' + delete: + tags: [Network Policy] + summary: Remove group entry + description: Remove an entry by entry_id or value from a policy group. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [kind, group_id] + properties: + kind: + type: string + enum: [channel_hashes, pubkeys] + group_id: + type: string + entry_id: + type: string + value: + type: string + responses: + '200': + description: Entry removed + content: + application/json: + schema: + $ref: '#/components/schemas/SuccessResponse' + # ============================================================================ # Identity Management # ============================================================================ diff --git a/tests/test_api_endpoints_core_coverage.py b/tests/test_api_endpoints_core_coverage.py index b55285d..0895324 100644 --- a/tests/test_api_endpoints_core_coverage.py +++ b/tests/test_api_endpoints_core_coverage.py @@ -1441,6 +1441,42 @@ def test_noise_floor_and_crc_endpoints(cherrypy_ctx): assert err["success"] is False +def test_metrics_graph_data_includes_policy_events(cherrypy_ctx): + del cherrypy_ctx + api = _make_api() + storage = SimpleNamespace( + get_rrd_data=MagicMock( + return_value={ + "start_time": 100, + "end_time": 220, + "step": 60, + "timestamps": [100, 160, 220], + "metrics": { + "rx_count": [10, 12, 15], + "tx_count": [8, 9, 11], + }, + } + ), + get_policy_event_counts=MagicMock( + return_value=[ + {"timestamp": 60, "count": 1}, + {"timestamp": 120, "count": 3}, + {"timestamp": 180, "count": 2}, + ] + ), + ) + _attach_storage(api, storage) + + out = api.metrics_graph_data(hours="24", resolution="average", metrics="rx_count,policy_events") + assert out["success"] is True + + series_by_type = {item["type"]: item for item in out["data"]["series"]} + assert "rx_count" in series_by_type + assert "policy_events" in series_by_type + assert series_by_type["policy_events"]["name"] == "Policy Events" + assert series_by_type["policy_events"]["data"] == [[100000, 1], [160000, 3], [220000, 2]] + + def test_advert_contact_and_rate_limit_stats_endpoints(cherrypy_ctx): del cherrypy_ctx api = _make_api() diff --git a/tests/test_api_policy_endpoints.py b/tests/test_api_policy_endpoints.py new file mode 100644 index 0000000..bdb983a --- /dev/null +++ b/tests/test_api_policy_endpoints.py @@ -0,0 +1,253 @@ +from types import SimpleNamespace + +import cherrypy +import yaml + +from repeater.web.api_endpoints import APIEndpoints + + +def _make_api(config=None, config_path="/tmp/config.yaml", daemon=None): + api = APIEndpoints.__new__(APIEndpoints) + api.config = config or {} + api.daemon_instance = daemon + api.send_advert_func = None + api.event_loop = None + api.stats_getter = None + api._config_path = config_path + api.config_manager = None + return api + + +def _set_request(monkeypatch, method="GET", payload=None): + request = SimpleNamespace(method=method, params={}, json=payload or {}) + response = SimpleNamespace(headers={}, status=200) + monkeypatch.setattr(cherrypy, "request", request, raising=False) + monkeypatch.setattr(cherrypy, "response", response, raising=False) + return request, response + + +def test_policy_get_returns_defaults_when_file_missing(tmp_path, monkeypatch): + cfg_path = tmp_path / "config.yaml" + cfg_path.write_text("repeater: {node_name: test}\n", encoding="utf-8") + api = _make_api(config={}, config_path=str(cfg_path)) + + _set_request(monkeypatch, method="GET") + result = api.policy() + + assert result["success"] is True + assert result["data"]["exists"] is False + assert result["data"]["policy_engine"]["enabled"] is False + assert result["data"]["policy_file"].endswith("policy.yaml") + + +def test_policy_post_saves_file_and_applies_runtime(tmp_path, monkeypatch): + cfg_path = tmp_path / "config.yaml" + cfg_path.write_text("repeater: {node_name: test}\n", encoding="utf-8") + + repeater_handler = SimpleNamespace(policy_engine=None) + daemon = SimpleNamespace(repeater_handler=repeater_handler) + api = _make_api(config={}, config_path=str(cfg_path), daemon=daemon) + + payload = { + "policy_engine": { + "enabled": True, + "default_action": "allow", + "rules": [ + { + "id": 10, + "enabled": True, + "if": { + "all": [ + { + "field": "hop_count", + "op": "greater_than", + "value": 4, + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + } + _set_request(monkeypatch, method="POST", payload=payload) + + result = api.policy() + + assert result["success"] is True + assert result["restart_required"] is False + assert result["data"]["policy_engine"]["enabled"] is True + + policy_path = tmp_path / "policy.yaml" + assert policy_path.exists() + + loaded = yaml.safe_load(policy_path.read_text(encoding="utf-8")) + assert loaded["policy_engine"]["enabled"] is True + assert len(loaded["policy_engine"]["rules"]) == 1 + + assert api.config["policy_engine"]["enabled"] is True + assert repeater_handler.policy_engine is not None + + +def test_policy_validate_returns_normalized_payload(tmp_path, monkeypatch): + cfg_path = tmp_path / "config.yaml" + cfg_path.write_text("repeater: {node_name: test}\n", encoding="utf-8") + api = _make_api(config={}, config_path=str(cfg_path)) + + payload = { + "enabled": 1, + "default_action": "allow", + "rules": [{"id": 1, "enabled": True, "if": {"all": []}, "then": {"action": "drop"}}], + } + _set_request(monkeypatch, method="POST", payload=payload) + + result = api.policy_validate() + + assert result["success"] is True + assert result["data"]["valid"] is True + assert result["data"]["normalized"]["enabled"] is True + assert result["data"]["effective"]["rule_count"] == 1 + + +def test_policy_groups_create_and_add_channel_hash_entries(tmp_path, monkeypatch): + cfg_path = tmp_path / "config.yaml" + cfg_path.write_text("repeater: {node_name: test}\n", encoding="utf-8") + api = _make_api(config={}, config_path=str(cfg_path)) + + _set_request( + monkeypatch, + method="POST", + payload={ + "kind": "channel_hashes", + "group_id": "ops_channels", + "friendly_name": "Ops Channels", + "description": "Operational channel hash group", + }, + ) + group_create = api.policy_groups() + assert group_create["success"] is True + assert group_create["data"]["group"]["friendly_name"] == "Ops Channels" + + _set_request( + monkeypatch, + method="POST", + payload={ + "kind": "channel_hashes", + "group_id": "ops_channels", + "value": "0x9CD8FCF22A47333B591D96A2B848B73F", + "friendly_name": "Ops Primary", + }, + ) + entry_create = api.policy_group_entries() + assert entry_create["success"] is True + assert entry_create["data"]["entry"]["value"] == "0x9CD8FCF22A47333B591D96A2B848B73F" + assert entry_create["data"]["entry"]["friendly_name"] == "Ops Primary" + + policy_path = tmp_path / "policy.yaml" + loaded = yaml.safe_load(policy_path.read_text(encoding="utf-8")) + groups = loaded["groups"] + assert groups["channel_hashes"][0]["id"] == "ops_channels" + assert ( + groups["channel_hashes"][0]["entries"][0]["value"] == "0x9CD8FCF22A47333B591D96A2B848B73F" + ) + + projected = loaded["policy_engine"]["objects"]["channel_hash_groups"] + assert projected["ops_channels"] == ["0x9CD8FCF22A47333B591D96A2B848B73F"] + + +def test_policy_groups_delete_accepts_query_params(tmp_path, monkeypatch): + cfg_path = tmp_path / "config.yaml" + cfg_path.write_text("repeater: {node_name: test}\n", encoding="utf-8") + api = _make_api(config={}, config_path=str(cfg_path)) + + _set_request( + monkeypatch, + method="POST", + payload={ + "kind": "channel_hashes", + "group_id": "ops_channels", + "friendly_name": "Ops Channels", + }, + ) + assert api.policy_groups()["success"] is True + + request, _ = _set_request(monkeypatch, method="DELETE", payload={}) + request.params = {"kind": "channel_hashes", "group_id": "ops_channels"} + + result = api.policy_groups() + + assert result["success"] is True + assert result["data"]["group_id"] == "ops_channels" + + +def test_policy_groups_create_and_add_pubkey_entries(tmp_path, monkeypatch): + cfg_path = tmp_path / "config.yaml" + cfg_path.write_text("repeater: {node_name: test}\n", encoding="utf-8") + api = _make_api(config={}, config_path=str(cfg_path)) + + _set_request( + monkeypatch, + method="POST", + payload={ + "kind": "pubkeys", + "group_id": "trusted_relays", + "friendly_name": "Trusted Relays", + }, + ) + group_create = api.policy_groups() + assert group_create["success"] is True + + _set_request( + monkeypatch, + method="POST", + payload={ + "kind": "pubkeys", + "group_id": "trusted_relays", + "value": "aabbccdd", + "friendly_name": "Relay Alpha", + }, + ) + entry_create = api.policy_group_entries() + assert entry_create["success"] is True + assert entry_create["data"]["entry"]["value"] == "0xaabbccdd" + assert entry_create["data"]["entry"]["friendly_name"] == "Relay Alpha" + + _set_request(monkeypatch, method="GET") + policy_get = api.policy() + assert policy_get["success"] is True + pubkey_groups = policy_get["data"]["groups"]["pubkeys"] + assert len(pubkey_groups) == 1 + assert pubkey_groups[0]["friendly_name"] == "Trusted Relays" + assert pubkey_groups[0]["entries"][0]["friendly_name"] == "Relay Alpha" + + +def test_policy_post_preserves_existing_groups(tmp_path, monkeypatch): + cfg_path = tmp_path / "config.yaml" + cfg_path.write_text("repeater: {node_name: test}\n", encoding="utf-8") + api = _make_api(config={}, config_path=str(cfg_path)) + + _set_request( + monkeypatch, + method="POST", + payload={ + "kind": "channel_hashes", + "group_id": "group_one", + "friendly_name": "Group One", + }, + ) + create_group = api.policy_groups() + assert create_group["success"] is True + + _set_request( + monkeypatch, + method="POST", + payload={ + "enabled": True, + "default_action": "allow", + "rules": [], + }, + ) + update_policy = api.policy() + assert update_policy["success"] is True + assert update_policy["data"]["policy_engine"]["enabled"] is True + assert len(update_policy["data"]["groups"]["channel_hashes"]) == 1 diff --git a/tests/test_packet_router.py b/tests/test_packet_router.py index 888abe9..b6aaf94 100644 --- a/tests/test_packet_router.py +++ b/tests/test_packet_router.py @@ -37,6 +37,7 @@ from repeater.packet_router import ( _companion_dedup_key, _is_direct_final_hop, ) +from repeater.policy_engine import PolicyEngine # --------------------------------------------------------------------------- # Minimal daemon stub @@ -80,6 +81,45 @@ def _make_bridge(): return bridge +class _SlottedPacket: + __slots__ = ( + "payload", + "header", + "rssi", + "snr", + "timestamp", + "_injected_for_tx", + "path", + "calculate_packet_hash", + "mark_do_not_retransmit", + "_payload_type", + ) + + def __init__(self, payload_type: int = 1): + self.payload = b"\x00" + self.header = 0x00 + self.rssi = -80 + self.snr = 5.0 + self.timestamp = time.time() + self._injected_for_tx = False + self.path = bytearray() + self.calculate_packet_hash = MagicMock(return_value=b"\x01" * 32) + self.mark_do_not_retransmit = MagicMock() + self._payload_type = payload_type + + def get_payload_type(self): + return self._payload_type + + def get_path_hash_size(self): + return 0 + + def get_path_hash_count(self): + return 0 + + def get_path_hashes_hex(self): + return [] + + # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- @@ -201,6 +241,44 @@ class TestInFlightCap(unittest.IsolatedAsyncioTestCase): finally: await router.stop() + async def test_policy_companion_precheck_handles_slotted_packet(self): + """Policy companion precheck must not attach attributes to slotted Packet objects.""" + daemon = _make_daemon() + daemon.companion_bridges = {"bridge": _make_bridge()} + daemon.repeater_handler.policy_engine = PolicyEngine({"enabled": True, "rules": []}) + router = PacketRouter(daemon) + pkt = _SlottedPacket(payload_type=1) + metadata = {"rssi": pkt.rssi, "snr": pkt.snr} + + bridges = router._companion_bridges_for_packet(pkt, metadata) + + self.assertEqual(bridges, daemon.companion_bridges) + self.assertIn("_policy_precheck_decision", metadata) + + async def test_route_grp_txt_reuses_policy_precheck_metadata(self): + """GRP_TXT should not force a second policy evaluation when the router already pre-checked it.""" + daemon = _make_daemon() + daemon.repeater_handler = AsyncMock(return_value=True) + daemon.repeater_handler.storage = MagicMock() + daemon.repeater_handler.record_packet_only = MagicMock() + daemon.repeater_handler.policy_engine = PolicyEngine({"enabled": True, "rules": []}) + evaluate_spy = patch.object( + daemon.repeater_handler.policy_engine, + "evaluate", + wraps=daemon.repeater_handler.policy_engine.evaluate, + ) + bridge = _make_bridge() + daemon.companion_bridges = {0x01: bridge} + router = PacketRouter(daemon) + pkt = _make_packet(GroupTextHandler.payload_type()) + + with evaluate_spy as mock_evaluate: + await router._route_packet(pkt) + + self.assertEqual(mock_evaluate.call_count, 1) + bridge.process_received_packet.assert_awaited_once() + daemon.repeater_handler.assert_awaited_once() + async def test_non_injected_handler_false_is_logged(self): """Inbound packets should log when repeater_handler reports TX failure.""" daemon = _make_daemon() @@ -476,6 +554,27 @@ class TestPacketRouterRoutingBranches(unittest.IsolatedAsyncioTestCase): bridge.process_received_packet.assert_awaited_once() daemon.repeater_handler.assert_awaited_once() + async def test_route_advert_policy_drop_blocks_companion_delivery(self): + daemon = _make_daemon() + daemon.advert_helper = MagicMock() + daemon.advert_helper.process_advert_packet = AsyncMock() + daemon.repeater_handler.policy_engine = PolicyEngine( + { + "enabled": True, + "default_action": "drop", + "rules": [], + } + ) + bridge = _make_bridge() + daemon.companion_bridges = {0x42: bridge} + router = PacketRouter(daemon) + pkt = _make_packet(AdvertHandler.payload_type()) + + await router._route_packet(pkt) + + daemon.advert_helper.process_advert_packet.assert_awaited_once() + bridge.process_received_packet.assert_not_awaited() + async def test_route_login_server_to_companion_marks_processed(self): daemon = _make_daemon() bridge = _make_bridge() diff --git a/tests/test_policy_engine.py b/tests/test_policy_engine.py new file mode 100644 index 0000000..fbb6bdd --- /dev/null +++ b/tests/test_policy_engine.py @@ -0,0 +1,749 @@ +from unittest.mock import patch + +import yaml +from pymc_core.protocol.constants import PAYLOAD_TYPE_GRP_TXT +from pymc_core.protocol.identity import LocalIdentity +from pymc_core.protocol.packet_builder import PacketBuilder + +from repeater.policy_engine import PolicyEngine + + +class _DummyPacket: + def __init__( + self, + payload=b"\x01\x02", + path_hashes=None, + transport_codes=None, + payload_type=None, + ): + self.payload = bytearray(payload) + self.transport_codes = transport_codes or [0, 0] + self._path_hashes = path_hashes or [] + self._payload_type = payload_type + + def get_path_hashes_hex(self): + return self._path_hashes + + def get_payload_type(self): + return self._payload_type + + +def test_policy_engine_disabled_allows(): + engine = PolicyEngine({"enabled": False, "rules": []}) + pkt = _DummyPacket() + + decision = engine.evaluate(pkt, {"hop_count": 3}) + + assert decision.action == "allow" + assert decision.matched is False + + +def test_policy_engine_first_match_wins_drop(): + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "rules": [ + { + "id": 10, + "enabled": True, + "if": {"all": [{"field": "hop_count", "op": "greater_than", "value": 2}]}, + "then": {"action": "drop"}, + }, + { + "id": 20, + "enabled": True, + "if": {"all": [{"field": "hop_count", "op": "greater_than", "value": 1}]}, + "then": {"action": "allow"}, + }, + ], + } + ) + pkt = _DummyPacket() + + decision = engine.evaluate(pkt, {"hop_count": 3}) + + assert decision.action == "drop" + assert decision.matched is True + assert decision.rule_id == 10 + + +def test_policy_engine_default_action_applies_when_no_match(): + engine = PolicyEngine( + { + "enabled": True, + "default_action": "drop", + "rules": [ + { + "id": 1, + "enabled": True, + "if": {"all": [{"field": "route_type", "op": "equals", "value": 1}]}, + "then": {"action": "allow"}, + } + ], + } + ) + pkt = _DummyPacket() + + decision = engine.evaluate(pkt, {"route_type": 2}) + + assert decision.action == "drop" + assert decision.matched is False + + +def test_policy_engine_log_only_action_is_returned_when_rule_matches(): + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "rules": [ + { + "id": 11, + "enabled": True, + "if": {"all": [{"field": "route_type", "op": "equals", "value": 1}]}, + "then": {"action": "log_only"}, + } + ], + } + ) + pkt = _DummyPacket() + + decision = engine.evaluate(pkt, {"route_type": 1}) + + assert decision.matched is True + assert decision.action == "log_only" + + +def test_load_config_reads_sibling_policy_yaml(tmp_path): + from repeater.config import load_config + + config_path = tmp_path / "config.yaml" + policy_path = tmp_path / "policy.yaml" + + config_path.write_text( + yaml.safe_dump( + { + "repeater": { + "node_name": "test", + "security": { + "max_clients": 1, + "admin_password": "a", + "guest_password": "g", + "allow_read_only": False, + "jwt_secret": "x", + "jwt_expiry_minutes": 60, + }, + }, + "radio": {"frequency": 869618000, "bandwidth": 62500}, + } + ), + encoding="utf-8", + ) + + policy_path.write_text( + yaml.safe_dump( + { + "policy_engine": { + "enabled": True, + "default_action": "allow", + "rules": [ + { + "id": 1, + "enabled": True, + "if": { + "all": [ + { + "field": "hop_count", + "op": "greater_than", + "value": 4, + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + } + ), + encoding="utf-8", + ) + + with patch("repeater.config._load_or_create_identity_key", return_value=b"k" * 32): + cfg = load_config(str(config_path)) + + assert cfg["policy_engine"]["enabled"] is True + assert len(cfg["policy_engine"]["rules"]) == 1 + assert cfg["policy_file_path"].endswith("policy.yaml") + + +def test_load_config_missing_policy_yaml_uses_safe_defaults(tmp_path): + from repeater.config import load_config + + config_path = tmp_path / "config.yaml" + config_path.write_text( + yaml.safe_dump( + { + "repeater": { + "node_name": "test", + "security": { + "max_clients": 1, + "admin_password": "a", + "guest_password": "g", + "allow_read_only": False, + "jwt_secret": "x", + "jwt_expiry_minutes": 60, + }, + }, + "radio": {"frequency": 869618000, "bandwidth": 62500}, + } + ), + encoding="utf-8", + ) + + with patch("repeater.config._load_or_create_identity_key", return_value=b"k" * 32): + cfg = load_config(str(config_path)) + + assert cfg["policy_engine"]["enabled"] is False + assert cfg["policy_engine"]["rules"] == [] + + +def test_policy_engine_path_hash_intersects_group_object(): + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "objects": { + "channel_hash_groups": { + "ops_channels": ["0x42", "0xAA"], + } + }, + "rules": [ + { + "id": 101, + "enabled": True, + "if": { + "all": [ + { + "field": "path_hashes", + "op": "intersects", + "value": "@channel_hash_groups.ops_channels", + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + pkt = _DummyPacket(path_hashes=["0x10", "0x42"]) + + decision = engine.evaluate(pkt, {"hop_count": 2}) + + assert decision.matched is True + assert decision.action == "drop" + + +def test_policy_engine_in_operator_for_scalar_vs_group_list(): + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "objects": {"allow_modes": {"prod_modes": ["forward", "monitor"]}}, + "rules": [ + { + "id": 202, + "enabled": True, + "if": { + "all": [ + { + "field": "mode", + "op": "in", + "value": "@allow_modes.prod_modes", + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + pkt = _DummyPacket() + + decision = engine.evaluate(pkt, {"mode": "forward"}) + + assert decision.matched is True + assert decision.action == "drop" + + +def test_policy_engine_matches_decrypted_channel_message_body_from_policy_objects(): + channel_secret = (b"policy-channel-secret" + b"\x00" * 32)[:32].hex() + packet = PacketBuilder.create_group_datagram( + group_name="ops", + local_identity=LocalIdentity(), + message="hello mesh from channel", + sender_name="Alice", + channels_config=[{"name": "ops", "secret": channel_secret}], + ) + + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "objects": { + "channels": { + "ops": { + "secret": channel_secret, + } + } + }, + "rules": [ + { + "id": 303, + "enabled": True, + "if": { + "all": [ + { + "field": "channel_message_body", + "op": "contains", + "value": "hello mesh", + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + + decision = engine.evaluate(packet, {"payload_type": packet.get_payload_type()}) + + assert decision.matched is True + assert decision.action == "drop" + + +def test_policy_engine_path_hashes_intersects_normalized_literal_list(): + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "rules": [ + { + "id": 404, + "enabled": True, + "if": { + "all": [ + { + "field": "path_hashes", + "op": "intersects", + "value": ["0x002A", "00AA"], + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + pkt = _DummyPacket(path_hashes=["002A", "00AA"]) + + decision = engine.evaluate(pkt, {"hop_count": 2}) + + assert decision.matched is True + assert decision.action == "drop" + + +def test_policy_engine_path_hashes_do_not_match_different_byte_lengths(): + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "rules": [ + { + "id": 405, + "enabled": True, + "if": { + "all": [ + { + "field": "path_hashes", + "op": "contains", + "value": "0x42", + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + pkt = _DummyPacket(path_hashes=["0042"]) + + decision = engine.evaluate(pkt, {"hop_count": 2}) + + assert decision.matched is False + assert decision.action == "allow" + + +def test_policy_engine_path_hashes_reject_mixed_literal_byte_lengths(): + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "rules": [ + { + "id": 406, + "enabled": True, + "if": { + "all": [ + { + "field": "path_hashes", + "op": "intersects", + "value": ["42", "0042"], + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + pkt = _DummyPacket(path_hashes=["42"]) + + decision = engine.evaluate(pkt, {"hop_count": 2}) + + assert decision.matched is False + assert decision.action == "allow" + + +def test_policy_engine_channel_hash_in_group_object(): + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "objects": { + "channel_hash_groups": { + "ops_channels": ["0x42", "0x99"], + } + }, + "rules": [ + { + "id": 407, + "enabled": True, + "if": { + "all": [ + { + "field": "channel_hash", + "op": "in", + "value": "@channel_hash_groups.ops_channels", + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + pkt = _DummyPacket(payload=b"\x42\xaa\xbb\xcc", payload_type=PAYLOAD_TYPE_GRP_TXT) + + decision = engine.evaluate(pkt, {"hop_count": 2}) + + assert decision.matched is True + assert decision.action == "drop" + + +def test_policy_engine_channel_hash_rejects_oversized_literal(): + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "rules": [ + { + "id": 408, + "enabled": True, + "if": { + "all": [ + { + "field": "channel_hash", + "op": "equals", + "value": "0x8B3387E9C5CDE8000000000000000000", + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + pkt = _DummyPacket(payload=b"\x11\xaa\xbb\xcc", payload_type=PAYLOAD_TYPE_GRP_TXT) + + decision = engine.evaluate(pkt, {"hop_count": 2}) + + assert decision.matched is False + assert decision.action == "allow" + + +def test_policy_engine_channel_hash_accepts_full_secret_literal(): + public_secret = "8b3387e9c5cdea6ac9e5edbaa115cd72" + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "rules": [ + { + "id": 409, + "enabled": True, + "if": { + "all": [ + { + "field": "channel_hash", + "op": "equals", + "value": public_secret, + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + # Derived hash for the secret above is 0x11. + pkt = _DummyPacket(payload=b"\x11\xaa\xbb\xcc", payload_type=PAYLOAD_TYPE_GRP_TXT) + + decision = engine.evaluate(pkt, {"hop_count": 2}) + + assert decision.matched is True + assert decision.action == "drop" + + +def test_policy_engine_channel_decryptable_true_with_matching_secret(): + channel_secret = (b"policy-channel-secret" + b"\x00" * 32)[:32].hex() + packet = PacketBuilder.create_group_datagram( + group_name="ops", + local_identity=LocalIdentity(), + message="decryptable test", + sender_name="Alice", + channels_config=[{"name": "ops", "secret": channel_secret}], + ) + + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "objects": { + "channels": { + "ops": { + "secret": channel_secret, + } + } + }, + "rules": [ + { + "id": 410, + "enabled": True, + "if": { + "all": [ + { + "field": "channel_decryptable", + "op": "equals", + "value": True, + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + + decision = engine.evaluate(packet, {"payload_type": packet.get_payload_type()}) + + assert decision.matched is True + assert decision.action == "drop" + + +def test_policy_engine_channel_decryptable_false_with_non_matching_secret(): + packet_secret = (b"packet-channel-secret" + b"\x00" * 32)[:32].hex() + wrong_secret = (b"wrong-channel-secret" + b"\x00" * 32)[:32].hex() + packet = PacketBuilder.create_group_datagram( + group_name="ops", + local_identity=LocalIdentity(), + message="undecryptable test", + sender_name="Alice", + channels_config=[{"name": "ops", "secret": packet_secret}], + ) + + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "objects": { + "channels": { + "ops": { + "secret": wrong_secret, + } + } + }, + "rules": [ + { + "id": 411, + "enabled": True, + "if": { + "all": [ + { + "field": "channel_decryptable", + "op": "equals", + "value": False, + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + + decision = engine.evaluate(packet, {"payload_type": packet.get_payload_type()}) + + assert decision.matched is True + assert decision.action == "drop" + + +def test_policy_engine_channel_decryptable_accepts_channels_list_with_psk(): + channel_secret = (b"policy-channel-secret" + b"\x00" * 32)[:32].hex() + packet = PacketBuilder.create_group_datagram( + group_name="ops", + local_identity=LocalIdentity(), + message="list schema decryptable test", + sender_name="Alice", + channels_config=[{"name": "ops", "secret": channel_secret}], + ) + + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "objects": { + "channels": [ + { + "name": "ops", + "psk": channel_secret, + } + ] + }, + "rules": [ + { + "id": 412, + "enabled": True, + "if": { + "all": [ + { + "field": "channel_decryptable", + "op": "equals", + "value": True, + } + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + + decision = engine.evaluate(packet, {"payload_type": packet.get_payload_type()}) + + assert decision.matched is True + assert decision.action == "drop" + + +def test_policy_engine_channel_decryptable_uses_inline_channel_hash_secret_literal(): + channel_secret = (b"policy-channel-secret" + b"\x00" * 32)[:32].hex() + packet = PacketBuilder.create_group_datagram( + group_name="ops", + local_identity=LocalIdentity(), + message="inline secret decryptable test", + sender_name="Alice", + channels_config=[{"name": "ops", "secret": channel_secret}], + ) + + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "objects": {"channels": {}}, + "rules": [ + { + "id": 413, + "enabled": True, + "if": { + "all": [ + { + "field": "channel_hash", + "op": "equals", + "value": channel_secret, + }, + { + "field": "channel_decryptable", + "op": "equals", + "value": True, + }, + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + + decision = engine.evaluate(packet, {"payload_type": packet.get_payload_type()}) + + assert decision.matched is True + assert decision.action == "drop" + + +def test_policy_engine_channel_decryptable_uses_channel_hash_group_secret(): + channel_secret = (b"policy-channel-secret" + b"\x00" * 32)[:32].hex() + packet = PacketBuilder.create_group_datagram( + group_name="ops", + local_identity=LocalIdentity(), + message="group secret decryptable test", + sender_name="Alice", + channels_config=[{"name": "ops", "secret": channel_secret}], + ) + + engine = PolicyEngine( + { + "enabled": True, + "default_action": "allow", + "objects": { + "channels": {}, + "channel_hash_groups": { + "ops_channels": [channel_secret], + }, + }, + "rules": [ + { + "id": 414, + "enabled": True, + "if": { + "all": [ + { + "field": "channel_hash", + "op": "in", + "value": "@channel_hash_groups.ops_channels", + }, + { + "field": "channel_decryptable", + "op": "equals", + "value": True, + }, + ] + }, + "then": {"action": "drop"}, + } + ], + } + ) + + decision = engine.evaluate(packet, {"payload_type": packet.get_payload_type()}) + + assert decision.matched is True + assert decision.action == "drop"