C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// """ Feature flags synchronisation plugin (AV mode only). In IM360 mode the Go resident-agent handles feature-flag sync. In AV mode there is no resident-agent, so this plugin takes over. Periodically POSTs the local file checksum to the API and writes back any updated flags to ``/var/imunify360/feature_flags.json`` (legacy map ``{flag: true}`` on disk) and ``/var/imunify360/feature_flags`` (plain names, one per line). The POSTed checksum is over the canonical JSON **array** of enabled names, matching the correlation sync API—not over the on-disk map bytes. """ import asyncio import json import logging import os import urllib.error import urllib.request from defence360agent.contracts.config import Core from defence360agent.contracts.plugins import MessageSource from defence360agent.internals.feature_flags import ( FLAGS_PATH, FLAGS_PLAIN_PATH, enabled_flag_names_sorted, plain_text_payload_for_enabled_flags, serialize_feature_flags_file_payload, sync_checksum_hex_from_flags_file, sync_response_file_bytes, ) from defence360agent.internals.iaid import ( IAIDTokenError, IndependentAgentIDAPI, ) from defence360agent.utils import Scope, atomic_rewrite logger = logging.getLogger(__name__) _SYNC_URL = "/api/sync/v1/feature-flags" def _env_int(name: str, default: int) -> int: """Read an int env var tolerantly. A non-numeric value (empty string, typo, etc.) must NOT raise at import time — the plugin lives in the AV agent entry point and a bad env var would otherwise kill the whole agent. """ raw = os.environ.get(name) if not raw: return default try: return int(raw) except ValueError: logger.warning( "feature-flags: %s=%r is not an int, using default %d", name, raw, default, ) return default _TRUE_VALUES = frozenset({"1", "true", "yes", "on"}) _FALSE_VALUES = frozenset({"0", "false", "no", "off"}) def _env_bool(name: str, default: bool) -> bool: raw = os.environ.get(name) if not raw: return default normalized = raw.strip().lower() if normalized in _TRUE_VALUES: return True if normalized in _FALSE_VALUES: return False logger.warning( "feature-flags: %s=%r is not a bool, using default %s", name, raw, default, ) return default _SYNC_INTERVAL = _env_int("I360_FEATURE_FLAGS_SYNC_INTERVAL", 3600) _INITIAL_DELAY = _env_int("I360_FEATURE_FLAGS_INIT_DELAY", 10) _UNREGISTERED_DELAY = _env_int("I360_FEATURE_FLAGS_UNREG_DELAY", 30) _USE_SERVER_DELAY = _env_bool("I360_FEATURE_FLAGS_USE_SERVER_DELAY", True) _HTTP_TIMEOUT = 30 def _next_delay(server_delay: int) -> int: if _USE_SERVER_DELAY and server_delay > 0: return server_delay return _SYNC_INTERVAL class FeatureFlagsSync(MessageSource): SCOPE = Scope.AV async def create_source(self, loop, sink): self._loop = loop self._sink = sink self._task = loop.create_task(self._sync_loop()) async def shutdown(self): if self._task is not None: self._task.cancel() try: await self._task except asyncio.CancelledError: pass def _local_checksum(self) -> str: return sync_checksum_hex_from_flags_file(FLAGS_PATH) async def _sync_loop(self): await asyncio.sleep(_INITIAL_DELAY) while True: delay = _SYNC_INTERVAL try: if not IndependentAgentIDAPI.is_registered(): delay = _UNREGISTERED_DELAY else: delay = _next_delay(await self._do_sync()) except asyncio.CancelledError: raise except Exception: logger.warning("feature flags sync failed", exc_info=True) await asyncio.sleep(delay) async def _do_sync(self) -> int: try: token = await IndependentAgentIDAPI.get_token() except IAIDTokenError: logger.warning("no IAID token, skipping feature flags sync") return 0 loop = asyncio.get_event_loop() checksum = await loop.run_in_executor(None, self._local_checksum) payload = json.dumps({"checksum": checksum}).encode() base_url = os.getenv("I360_FEATURE_FLAGS_API_URL", Core.API_BASE_URL) url = base_url.rstrip("/") + _SYNC_URL req = urllib.request.Request( url, data=payload, headers={ "Content-Type": "application/json", "X-Auth": token, }, method="POST", ) try: resp_body = await loop.run_in_executor( None, self._blocking_request, req ) except urllib.error.HTTPError as e: # Non-5xx (404/403/4xx) is usually a server-side routing or # auth state, not an agent bug — keep it a one-line WARNING. # 5xx means the server actually misbehaved; keep the traceback. if 500 <= e.code < 600: logger.error( "feature flags sync HTTP %s on %s: %s", e.code, url, e.reason, ) else: logger.warning( "feature flags sync HTTP %s on %s: %s", e.code, url, e.reason, ) return 0 except (urllib.error.URLError, TimeoutError) as e: # DNS, connection refused, TLS, timeout — transient network # conditions, not bugs. One-line WARNING so logs stay readable. # A timeout during resp.read() escapes urlopen as a bare # TimeoutError, not wrapped in URLError. logger.warning( "feature flags sync connection failed on %s: %s", url, getattr(e, "reason", e), ) return 0 except Exception: logger.error( "feature flags sync request failed on %s", url, exc_info=True, ) return 0 try: result = json.loads(resp_body) except json.JSONDecodeError: logger.error("failed to parse feature flags response") return 0 server_delay = result.get("delay", 0) if result.get("changed") is False: logger.debug("feature flags unchanged, skipping write") return server_delay flags = result.get("flags") params = result.get("params") or {} if flags is not None: await loop.run_in_executor(None, self._write_flags, flags, params) return server_delay @staticmethod def _blocking_request(req: urllib.request.Request) -> bytes: with urllib.request.urlopen(req, timeout=_HTTP_TIMEOUT) as resp: return resp.read() @staticmethod def _write_flags(flags, params=None) -> None: """Persist flags + params on disk in the c