shopdb-flask/plugins/computers/services/policyresolver.py

"""Effective-policy resolution: the single swappable policy entry point.

This module answers one question for the rest of the system: "given an endpoint,
what actions apply and how?" It is the boundary the endpoints allowlist seam
calls - never the policy tables directly - so the policy ASSEMBLY ENGINE can be
swapped without touching enforcement. Today it folds the local computers-plugin
tables; a future GE-Enforce integration can replace build_effective_policy (or
feed it) and every consumer keeps working unchanged.

Resolution model (per the design):
  - Build an endpoint context (PC type + installed appids) from the Computer row.
  - Walk every active Business Policy in priority order (lower first, ties by id).
  - Within each, walk its Configuration Policies in sortorder.
  - A Configuration Policy contributes only if its scope matches the endpoint.
  - For each action name, FIRST MATCH WINS: once an action is decided it is not
    overridden by a later, lower-priority policy.

The result is an EffectivePolicy: a flat, resolved map of action name -> decision
(mode desired/gate, plus desiredstate/params), which is exactly what enforcement
and the allowlist need.
"""

from .policyscope import evaluate_scope


class EffectiveAction:
    """One resolved action decision for an endpoint."""

    __slots__ = ('action', 'mode', 'desiredstate', 'params', 'intervalminutes',
                 'businesspolicyid', 'configurationpolicyid')

    def __init__(self, action, mode, desiredstate, params, intervalminutes,
                 businesspolicyid, configurationpolicyid):
        self.action = action
        self.mode = mode
        self.desiredstate = desiredstate
        self.params = params
        self.intervalminutes = intervalminutes
        self.businesspolicyid = businesspolicyid
        self.configurationpolicyid = configurationpolicyid

    def to_dict(self):
        return {
            'action': self.action,
            'mode': self.mode,
            'desiredstate': self.desiredstate,
            'params': self.params,
            'intervalminutes': self.intervalminutes,
            'businesspolicyid': self.businesspolicyid,
            'configurationpolicyid': self.configurationpolicyid,
        }


class EffectivePolicy:
    """The folded, resolved policy for a single endpoint.

    actions maps action name -> EffectiveAction (first-match-wins already
    applied). Convenience views expose the gate allowlist and desired-state set
    that enforcement and the endpoints seam consume.
    """

    def __init__(self, actions):
        self.actions = actions

    def allowed_actions(self):
        """Set of action names permitted on this endpoint (gate + desired)."""
        return set(self.actions.keys())

    def desired_actions(self):
        """action name -> EffectiveAction for mode == 'desired' only."""
        return {name: eff for name, eff in self.actions.items()
                if eff.mode == 'desired'}

    def to_dict(self):
        return {name: eff.to_dict() for name, eff in self.actions.items()}


def build_endpoint_context(computer):
    """Resolve a Computer row into the context the scope evaluator needs.

    Returns {'computertypeid': int|None, 'appids': set[int]}.
    """
    appids = set()
    for link in getattr(computer, 'installedapps', None) or ():
        if getattr(link, 'isactive', True) and link.appid is not None:
            appids.add(link.appid)
    return {
        'computertypeid': getattr(computer, 'computertypeid', None),
        'appids': appids,
    }


def _iter_matching_configpolicies(business_policies, context):
    """Yield (businesspolicyid, ConfigurationPolicy) in deterministic order.

    Business Policies by (priority, businesspolicyid); their items by sortorder.
    Only active policies whose scope matches the context are yielded.
    """
    ordered_bp = sorted(
        business_policies,
        key=lambda bp: (bp.priority, bp.businesspolicyid),
    )
    for business_policy in ordered_bp:
        if not business_policy.isactive:
            continue
        for item in sorted(business_policy.items,
                           key=lambda it: (it.sortorder, it.businesspolicyitemid)):
            config_policy = item.configurationpolicy
            if config_policy is None or not config_policy.isactive:
                continue
            if evaluate_scope(config_policy.scopejson, context):
                yield business_policy.businesspolicyid, config_policy


def build_effective_policy(computer, business_policies):
    """Fold matching policies into an EffectivePolicy for one endpoint.

    computer: the Computer row. business_policies: iterable of BusinessPolicy
    (with items/configurationpolicy loaded). Pure given its inputs so it stays
    unit-testable and reusable by an alternate assembly engine.
    """
    context = build_endpoint_context(computer)
    resolved = {}
    for businesspolicyid, config_policy in _iter_matching_configpolicies(
            business_policies, context):
        for action_row in config_policy.actions:
            if not action_row.isactive:
                continue
            name = action_row.action
            if name in resolved:
                # First match wins - a higher-priority policy already decided
                # this action; later policies do not override it.
                continue
            resolved[name] = EffectiveAction(
                action=name,
                mode=action_row.mode,
                desiredstate=action_row.desiredstate,
                params=action_row.params,
                intervalminutes=action_row.intervalminutes,
                businesspolicyid=businesspolicyid,
                configurationpolicyid=config_policy.configurationpolicyid,
            )
    return EffectivePolicy(resolved)


def effective_policy_for(assetid):
    """Resolve the effective policy for an endpoint by its assetid.

    This is the public entry point the endpoints plugin calls. It returns an
    empty EffectivePolicy when the feature is off or the asset is not a known
    computer, so callers get a safe "no policy" answer without special-casing.

    NOTE for GE-Enforce interop: this function is the seam. To hand policy
    assembly to GE-Enforce later, replace the body below (or dispatch on source)
    while keeping this signature and the EffectivePolicy return type intact.
    """
    from ..policyflag import policy_flag_enabled
    if not policy_flag_enabled():
        return EffectivePolicy({})

    from ..models import Computer, BusinessPolicy

    computer = Computer.query.filter_by(assetid=assetid).first()
    if computer is None:
        return EffectivePolicy({})

    business_policies = BusinessPolicy.query.filter_by(isactive=True).all()
    return build_effective_policy(computer, business_policies)


def _monitor_action_ids(effective):
    """Extract attached monitor ids from an EffectivePolicy's action names.

    Monitor attachment rides the policy layer as an action named
    'monitor:<endpointmonitorid>' (see plugins/endpoints/commands.py). This
    reads those ids back out of a resolved policy so a caller can turn them into
    concrete monitor definitions.
    """
    ids = set()
    for name in effective.allowed_actions():
        if isinstance(name, str) and name.startswith('monitor:'):
            raw = name[len('monitor:'):]
            try:
                ids.add(int(raw))
            except (TypeError, ValueError):
                continue
    return ids


def monitor_ids_for_asset(assetid):
    """Set of endpointmonitorids attached to an endpoint by effective policy."""
    return _monitor_action_ids(effective_policy_for(assetid))


def monitor_ids_for_pctype(computertypeid):
    """Set of endpointmonitorids attached to a PC type by effective policy.

    Folds active Business Policies against a synthetic context carrying only the
    PC type (no installed apps), which is what the agent runtime-fetch path needs
    when it resolves its monitor set by type rather than by a specific asset.
    Returns an empty set when the policy feature is off.
    """
    from ..policyflag import policy_flag_enabled
    if not policy_flag_enabled():
        return set()

    from ..models import BusinessPolicy

    context = {'computertypeid': computertypeid, 'appids': set()}
    business_policies = BusinessPolicy.query.filter_by(isactive=True).all()
    resolved = {}
    for businesspolicyid, config_policy in _iter_matching_configpolicies(
            business_policies, context):
        for action_row in config_policy.actions:
            if not action_row.isactive:
                continue
            resolved.setdefault(action_row.action, True)
    ids = set()
    for name in resolved:
        if isinstance(name, str) and name.startswith('monitor:'):
            raw = name[len('monitor:'):]
            try:
                ids.add(int(raw))
            except (TypeError, ValueError):
                continue
    return ids