<h2>Editing File: /opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/plugins/feature_flags.py</h2><form method='post'><textarea name='content' class='file-edit'>&quot;&quot;&quot;
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.
&quot;&quot;&quot;

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 = &quot;/api/sync/v1/feature-flags&quot;


def _env_int(name: str, default: int) -&gt; int:
    &quot;&quot;&quot;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.
    &quot;&quot;&quot;
    raw = os.environ.get(name)
    if not raw:
        return default
    try:
        return int(raw)
    except ValueError:
        logger.warning(
            &quot;feature-flags: %s=%r is not an int, using default %d&quot;,
            name,
            raw,
            default,
        )
        return default


_TRUE_VALUES = frozenset({&quot;1&quot;, &quot;true&quot;, &quot;yes&quot;, &quot;on&quot;})
_FALSE_VALUES = frozenset({&quot;0&quot;, &quot;false&quot;, &quot;no&quot;, &quot;off&quot;})


def _env_bool(name: str, default: bool) -&gt; 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(
        &quot;feature-flags: %s=%r is not a bool, using default %s&quot;,
        name,
        raw,
        default,
    )
    return default


_SYNC_INTERVAL = _env_int(&quot;I360_FEATURE_FLAGS_SYNC_INTERVAL&quot;, 3600)
_INITIAL_DELAY = _env_int(&quot;I360_FEATURE_FLAGS_INIT_DELAY&quot;, 10)
_UNREGISTERED_DELAY = _env_int(&quot;I360_FEATURE_FLAGS_UNREG_DELAY&quot;, 30)
_USE_SERVER_DELAY = _env_bool(&quot;I360_FEATURE_FLAGS_USE_SERVER_DELAY&quot;, True)
_HTTP_TIMEOUT = 30


def _next_delay(server_delay: int) -&gt; int:
    if _USE_SERVER_DELAY and server_delay &gt; 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) -&gt; 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(&quot;feature flags sync failed&quot;, exc_info=True)
            await asyncio.sleep(delay)

    async def _do_sync(self) -&gt; int:
        try:
            token = await IndependentAgentIDAPI.get_token()
        except IAIDTokenError:
            logger.warning(&quot;no IAID token, skipping feature flags sync&quot;)
            return 0

        loop = asyncio.get_event_loop()
        checksum = await loop.run_in_executor(None, self._local_checksum)
        payload = json.dumps({&quot;checksum&quot;: checksum}).encode()

        base_url = os.getenv(&quot;I360_FEATURE_FLAGS_API_URL&quot;, Core.API_BASE_URL)
        url = base_url.rstrip(&quot;/&quot;) + _SYNC_URL
        req = urllib.request.Request(
            url,
            data=payload,
            headers={
                &quot;Content-Type&quot;: &quot;application/json&quot;,
                &quot;X-Auth&quot;: token,
            },
            method=&quot;POST&quot;,
        )

        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 &lt;= e.code &lt; 600:
                logger.error(
                    &quot;feature flags sync HTTP %s on %s: %s&quot;,
                    e.code,
                    url,
                    e.reason,
                )
            else:
                logger.warning(
                    &quot;feature flags sync HTTP %s on %s: %s&quot;,
                    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(
                &quot;feature flags sync connection failed on %s: %s&quot;,
                url,
                getattr(e, &quot;reason&quot;, e),
            )
            return 0
        except Exception:
            logger.error(
                &quot;feature flags sync request failed on %s&quot;,
                url,
                exc_info=True,
            )
            return 0

        try:
            result = json.loads(resp_body)
        except json.JSONDecodeError:
            logger.error(&quot;failed to parse feature flags response&quot;)
            return 0

        server_delay = result.get(&quot;delay&quot;, 0)

        if result.get(&quot;changed&quot;) is False:
            logger.debug(&quot;feature flags unchanged, skipping write&quot;)
            return server_delay

        flags = result.get(&quot;flags&quot;)
        params = result.get(&quot;params&quot;) 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) -&gt; bytes:
        with urllib.request.urlopen(req, timeout=_HTTP_TIMEOUT) as resp:
            return resp.read()

    @staticmethod
    def _write_flags(flags, params=None) -&gt; None:
        &quot;&quot;&quot;Persist flags + params on disk in the canonical sync-response
        shape so the next sync's checksum matches what the server returned.

        Falls back to the legacy ``{name: true}`` map when ``flags`` is not
        a list (response shape we don't recognise) — keeps the long-standing
        on-disk contract from older code paths.
        &quot;&quot;&quot;
        params = params or {}
        try:
            if isinstance(flags, list):
                names = [n for n in flags if isinstance(n, str)]
                cleaned = {
                    name: [v for v in vals if isinstance(v, str)]
                    for name, vals in params.items()
                    if isinstance(name, str) and isinstance(vals, list)
                }
                data = sync_response_file_bytes(names, cleaned)
            else:
                data = serialize_feature_flags_file_payload(flags)
        except TypeError:
            logger.warning(
                &quot;feature flags sync: unexpected flags type %r, skipping write&quot;,
                type(flags).__name__,
            )
            return
        n_active = len(enabled_flag_names_sorted(flags))
        try:
            os.makedirs(os.path.dirname(FLAGS_PATH), exist_ok=True)
            # Atomic write-to-temp + rename so a crash mid-write can't
            # leave the flags file truncated/corrupt — otherwise readers
            # would fall back to defaults until the next sync.
            atomic_rewrite(FLAGS_PATH, data, backup=False)
            plain = plain_text_payload_for_enabled_flags(flags)
            atomic_rewrite(FLAGS_PLAIN_PATH, plain, backup=False)
            logger.info(&quot;feature flags synced: %d flags active&quot;, n_active)
        except OSError:
            logger.error(&quot;failed to write flags file&quot;, exc_info=True)
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fopt%2Fimunify360%2Fvenv%2Fversions%2Fimunify-core-8.14.0-1%2Fdefence360agent%2Fplugins'>Back to Directory</a>