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

@@ -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)$

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:

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
 
 

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()

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)
 

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)

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)

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

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(

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:

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
   # ============================================================================

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()

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

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()

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"