golden hour

/opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/feature_management/rpc/endpoints

File/Folder	Size	Actions
__init__.py	75 B
__pycache__	-
native.py	1.47 KB
show.py	2.55 KB
update.py	2.24 KB
utils.py	1.26 KB

Edit: update.py

from logging import getLogger
from typing import Any, List

import defence360agent.subsys.panels.hosting_panel as hp
from defence360agent.contracts.config import ConfigFile
from defence360agent.feature_management.constants import (
    AV,
    AV_REPORT,
    FULL,
    LOG,
    NA,
    PROACTIVE,
)
from defence360agent.feature_management.rpc.endpoints.utils import (
    builtin_feature_management_only,
)
from defence360agent.feature_management.utils import (
    update_default,
    update_users,
)
from defence360agent.rpc_tools.lookup import RootEndpoints, bind

logger = getLogger(__name__)

def _proactive_disable_target() -> str:
    """Return the permission value that "disable proactive" should resolve to.

Reads FEATURE_MANAGEMENT.proactive_disable_target. Default "na"
    preserves legacy behavior; "log" gives observe-only PROACTIVE_DEFENCE
    instead of fully off.
    """
    value = ConfigFile().get("FEATURE_MANAGEMENT", "proactive_disable_target")
    return LOG if value == "log" else NA

@builtin_feature_management_only
class FeatureManagementUpdateEndpoints(RootEndpoints):
    @staticmethod
    async def _update(feature: str, users: List[str], value: Any):
        if not users:
            return {
                "items": "succeed"
                if await update_default(feature, value)
                else "failed"
            }
        return {
            "items": await update_users(
                feature, users, value, await hp.HostingPanel().get_users()
            )
        }

def _adapt_value(self, feature, value):
        if value:
            return FULL
        if feature == AV:
            return AV_REPORT
        if feature == PROACTIVE:
            return _proactive_disable_target()
        return NA

@bind("feature-management", "enable")
    async def feature_management_enable(self, feature: str, users=None):
        """Enable specified feature"""
        return await self._update(
            feature, users, self._adapt_value(feature, True)
        )

@bind("feature-management", "disable")
    async def feature_management_disable(self, feature: str, users=None):
        """Disable specified feature"""
        return await self._update(
            feature, users, self._adapt_value(feature, False)
        )