FM / customer_local

/var/opt/nydus/ops/customer_local_ops

Name	Size	Mode	Actions
control_panel/	-	0755	rm
openclaw/	-	0755	rm
operating_system/	-	0755	rm
util/	-	0755	rm
__pycache__/	-	0755	rm
exceptions.py	476	0644	edit dl rm
__init__.py	35184	0644	edit dl rm

Edit: /var/opt/nydus/ops/customer_local_ops/__init__.py (35184B)

from datetime import datetime, timedelta
from typing import Any, Dict, List, Tuple, Union
from enum import Enum
import importlib
import json
import logging
import shutil
import sys as _sys
import os as _os

from customer_local_ops.util.constants import CANONICAL_TIMESTRING_FORMAT
from customer_local_ops.util.encryptor import Encryptor
from customer_local_ops.exceptions import EncryptionKeyNotFound
from customer_local_ops.util.execute import runCommand
from customer_local_ops.util.helpers import create_file
from customer_local_ops.util.retry import Retry
from .util import constants as _constants
from .util import encryptor as _encryptor
from .util import sizes as _sizes

__path__.append(_os.path.join(_os.path.dirname(__file__), 'util'))  # type: ignore
_sys.modules[__name__ + '.constants'] = _constants
_sys.modules[__name__ + '.encryptor'] = _encryptor
_sys.modules[__name__ + '.sizes'] = _sizes
constants = _constants
encryptor = _encryptor
sizes = _sizes

LOG = logging.getLogger(__name__)

class OpType(Enum):
    CONTROL_PANEL = "cp_op"
    OPERATING_SYSTEM = "os_op"
    CONTROL_PANEL_OPERATING_SYSTEM = "cp_os_op"
    UNKNOWN = "unknown"

# TODO: Move to central repo
class ResourceType(Enum):
    OVH = "ovh"
    OPENSTACK = "openstack"
    VIRTUOZZO_VM = "virtuozzo_vm"
    OPENSTACK_HOSTINGCLOUD = "openstack_hostingcloud"

RESOURCE_TYPES = list(ResourceType)

NydusResult = Union[Retry, Tuple[bool, Dict[str, Any]], Tuple[bool, Dict[str, Any], int]]

class OpResult:
    """
    Class whose objects encapsulate the complete result of an operation.
    """
    def __init__(self, success: bool = False, result: Dict[str, Any] = None, retry_interval: int = None):
        """
        Initialize an OpResult instance

:param success: Indicates whether or not the operation was successful
        :param result: A dictionary containing any messages or other data from the operation
        :param retry_interval:  An optional retry interval, in seconds. If this value is greater that or equal to
                                zero, the operation will be retried in that many seconds.
        """
        self.success = success
        self.result = result
        self.retry_interval = retry_interval

@property
    def errs(self) -> str:
        """Convenience property for accessing the 'errs` key in the result dict"""
        return self.result.get('errs', '') if isinstance(self.result, dict) else ''

@property
    def outs(self) -> str:
        """Convenience property for accessing the 'outs` key in the result dict"""
        return self.result.get('outs', '') if isinstance(self.result, dict) else ''

def as_tuple(self) -> NydusResult:
        """
        Construct and return a correctly-sized tuple, based on whether or not a retry_interval value is present, for
        passing back to Nydus.
        """
        return ((self.success, self.result) if self.retry_interval is None
                else (self.success, self.result, self.retry_interval))

class Ops:
    DISK_UTILIZATION_PATH = None
    RETRYABLE_ERRS = []
    PANOPTA_MANIFEST_FILE = '/etc/panopta-agent-manifest'
    DISTRO = None  # override
    OS_VERSION = None  # override
    QEMU_PACKAGE_NAME = None  # override

def __str__(self):
        return self.__class__.__name__

op_type = OpType.UNKNOWN

def get_op_type(self):
        return self.op_type

def _get_vm_tag(self):
        """Return this VM's tag.  Same return format as runCommand."""
        raise NotImplementedError

def _encryption_key(self):
        """The encryption key."""
        exit_code, outs, errs = self._get_vm_tag()
        if exit_code != 0 or not outs:
            LOG.error('Could not obtain encryption key')
            raise EncryptionKeyNotFound(exit_code, outs, errs)
        return outs

def decrypt(self, encrypted):
        """Decrypt an encrypted value."""
        try:
            key = self._encryption_key()
            return Encryptor.decrypt(encrypted, key)
        except Exception:
            LOG.exception('Decrypt failed')
            raise

def encrypt(self, target: str) -> str:
        """
        Encrypt a plain-text value.

:param target: The value to be encrypted
        :return: An encrypted string
        """
        try:
            key = self._encryption_key()
            return Encryptor.encrypt(target, key)
        except Exception:
            LOG.exception('Encrypt failed')
            raise

def write_file(self, path, content):
        LOG.debug('Writing file: %s', path)
        with open(path, 'w', encoding='utf-8') as f:
            f.write(content)

def get_os_op(self, os_op: str, control_panel_class_name: str = None) -> "Ops":
        """
        Get an instance of an operating system Ops class or ControlPanel-OS Ops class

:param os_op: The operating system class name e.g. linux.CentOS7
        :param control_panel_class_name: The control panel class name e.g. cpanel.CPanel
        :return: An instance of the operating system class e.g. CentOS7 or CP-OS class e.g. CentOS7CPanel
        """
        parts = os_op.split(".")
        # Last part of os_op should be the family name followed by the os class name, e.g. linux.CentOS7
        os_name = parts[-1]
        family_name = parts[-2]
        prefix = "customer_local_ops"
        if control_panel_class_name is None and self.get_op_type() == OpType.CONTROL_PANEL:
            # If we're requesting to get the os op class from a control panel class,
            # assume we want the operating system/control panel hybrid class
            control_panel_class_name = str(self)
        if control_panel_class_name is not None:
            # Return the operating system/control panel hybrid class
            os_name = os_name + control_panel_class_name
            cp_class_name = control_panel_class_name.lower()
            module_path = "{prefix}.control_panel.{family_name}_{cp_class_name}".format(prefix=prefix,
                                                                                        family_name=family_name,
                                                                                        cp_class_name=cp_class_name)
        else:
            module_path = "{prefix}.operating_system.{family_name}".format(prefix=prefix, family_name=family_name)
        mod = importlib.import_module(module_path)
        class_ = getattr(mod, os_name)
        os_op = class_()
        return os_op

def build_result_dict(self, outs: str, errs: str, op_name: str) -> Dict[str, Any]:
        """
        This function takes command output (outs, errs) and builds and returns a result
        dict with op metadata and errors if appropriate

:param outs: The stdout from the run command
        :param errs: The stderr from the run command
        :param op_name: the name of the op calling this function
        :return: a result dict
        """
        unicode_null = "\u0000"
        outs = outs.replace(unicode_null, "")
        errs = errs.replace(unicode_null, "")

result = {'outs': outs, 'errs': errs, 'cls_name': str(self), 'op_type': self.get_op_type().value,
                  'op_name': op_name}
        return result

def get_result_data(self, result: Any) -> OpResult:
        """
        This function takes a result tuple or result dict and parses it to a result tuple of a consistent format
        that can then be used by ops to inspect the result or build another result based on the first one

:param result: a dict or tuple result from an op
        :return: A data structure containing the complete result
        """
        if isinstance(result, tuple):
            result_len = len(result)
            if result_len == 1:
                return OpResult(True, result[0])
            if result_len == 2:
                return OpResult(result[0], result[1])
            if result_len == 3:
                return OpResult(result[0], result[1], result[2])
            # Should never get here
            raise ValueError("result tuple has too many elements: {}".format(result_len))
        return OpResult(True, result)

def build_result_from_other_result(self, result: Any, op_name: str) -> NydusResult:
        """
        This function takes a result tuple/dict/Retry/str from a secondary op (usually an os op) and (if not a Retry)
        extracts the metadata, output and errors from it and places it inside a result with metadata from the primary
        op (usually a control panel op).
        In this way it is clear:
        a). what the primary op was and
        b). that the error/output is coming from the secondary op

:param result: a dict or tuple result from an op
        :param op_name: the name of the op calling this function
        :return: a result dict, tuple or Retry instance
        """
        if isinstance(result, Retry):
            return result

data = self.get_result_data(result)
        if isinstance(data.result, dict):
            op_type = data.result.get('op_type', self.get_op_type().value)
            op_name_other = data.result.get('op_name', op_name)
            cls_name = data.result.get('cls_name', str(self))
            prefix = "{op_type} {cls_name}.{op_name_other}".format(op_type=op_type, cls_name=cls_name,
                                                                   op_name_other=op_name_other)
            outs = "{prefix} outs: {outs}".format(prefix=prefix, outs=data.outs)
            errs = data.errs
            if errs:
                errs = "{prefix} errs: {errs}".format(prefix=prefix, errs=data.errs)
        else:
            if data.success:
                outs = str(data.result)
                errs = ''
            else:
                outs = ''
                errs = str(data.result)
        data.result = self.build_result_dict(outs, errs, op_name)
        return data.as_tuple()

def build_result_from_cmd_output(self, exit_code: int, outs: str, errs: str, op_name: str,
                                     success_message: str = None) -> NydusResult:
        """
        This function takes a result from a command output (exit_code, outs, errs) and builds and returns a result
        tuple with op metadata and errors if appropriate

:param exit_code: The exit code returned from the run command
        :param outs: The stdout from the run command
        :param errs: The stderr from the run command
        :param op_name: the name of the op calling this function
        :param success_message: A message to use in place of 'outs' if the command ran successfully (optional)
        :return: a result tuple
        """
        success = exit_code == 0
        if success and success_message is not None:
            outs = success_message
        return success, self.build_result_dict(outs, errs, op_name)

def run_command_and_handle_result(self,
                                      command: Union[List[str], str],
                                      op_name: str,
                                      *args: Any,
                                      intermediate_result: Dict[str, Any] = None,
                                      **kw: Any) -> NydusResult:
        """Run command, return result suitable for a Nydus operation.

:param command: command list or string as per runCommand
        :param op_name: name of the operation; used as runCommand tag and result op_name
        :param args: additional positional arguments for runCommand
        :param intermediate_result:  Dict containing metadata for retries
        :param kw: additional keyword arguments for runCommand
        :return: command result as a Nydus operation result
        """
        exit_code, outs, errs = runCommand(command, op_name, *args, **kw)
        return self._result_handler(exit_code, outs, errs, op_name, intermediate_result=intermediate_result)

# def configure_ips(self, network_node):
    #     raise NotImplementedError

# def patch_system(self, payload):
    #     raise NotImplementedError

# def patch_system_specific_update(self, payload):
    #     raise NotImplementedError

# def list_vm_patches(self, payload):
    #     raise NotImplementedError

def add_user(self, payload, unused=None):
        raise NotImplementedError

# def remove_user(self, payload):
    #     raise NotImplementedError

# def change_password(self, payload):
    #     raise NotImplementedError

# def change_hostname(self, payload):
    #     raise NotImplementedError

# def install_package(self, payload):
    #     raise NotImplementedError

# def uninstall_package(self, payload):
    #     raise NotImplementedError

def shutdown_clean(self, unused=None):
        # The `unused` param is an artifact of Archon workflows requiring an I/O
        # chain for sequencing.
        raise NotImplementedError

# def stop_web_server(self, none):
    #     raise NotImplementedError

# def start_web_server(self, none):
    #     raise NotImplementedError

# def prepare_web_server_ssl(self, none):
    #     raise NotImplementedError

# def deploy_cert(self, payload):
    #     raise NotImplementedError

# def install_cert(self, payload):
    #     raise NotImplementedError

# def configure_mta(self, payload):
    #     raise NotImplementedError

def enable_admin(self, username, unused=None):
        # The `unused` param is an artifact of Archon workflows requiring an I/O
        # chain for sequencing.
        raise NotImplementedError

def disable_admin(self, username, unused=None):
        # The `unused` param is an artifact of Archon workflows requiring an I/O
        # chain for sequencing.
        raise NotImplementedError

def disable_all_admins(self, unused=None):
        # The `unused` param is an artifact of Archon workflows requiring an I/O
        # chain for sequencing.
        raise NotImplementedError

def extend_disk(self, payload, unused=None):
        # The `unused` param is an artifact of Archon workflows requiring an I/O
        # chain for sequencing.
        raise NotImplementedError

def _get_cpu_utilization(self) -> Dict[str, float]:
        """Return current CPU utilization.

:returns: dictionary with keys:
            - cpuUsed: percentage in range 0-100
        """
        raise NotImplementedError

def _get_disk_utilization(self) -> Dict[str, int]:
        """Return current disk utilization.

:returns: dictionary with keys:
            - diskTotal: mebibytes
            - diskUsed: mebibytes
        """
        disk = shutil.disk_usage(self.DISK_UTILIZATION_PATH)
        return {'diskTotal': disk.total >> 20,  # B > MiB
                'diskUsed': disk.used >> 20}

def _get_memory_utilization(self) -> Dict[str, int]:
        """Return current memory utilization.

:returns: dictionary with keys:
            - memoryTotal: mebibytes
            - memoryUsed: mebibytes
        """
        raise NotImplementedError

def get_utilization(self) -> Dict[str, Union[float, int, str]]:
        """Return current system utilization.

:returns: dictionary with keys:
            - collected: current time in
                :data:`~customer_local_ops.constants.CANONICAL_TIMESTRING_FORMAT`
            - memoryTotal: mebibytes
            - memoryUsed: mebibytes
            - diskTotal: mebibytes
            - diskUsed: mebibytes
            - cpuUsed: percentage in range 0-100
        """
        result = {}
        for stats in (self._get_cpu_utilization(),
                      self._get_disk_utilization(),
                      self._get_memory_utilization()):
            result.update(stats)
        result['collected'] = datetime.utcnow().strftime(
            CANONICAL_TIMESTRING_FORMAT)
        return result

def _create_panopta_manifest_file(self, payload):
        """ Panopta agent uses a manifest file for configuration
        of the customer id and templates to use.  This method creates that manifest file.

"""
        manifest_file = self.PANOPTA_MANIFEST_FILE
        customer_key = payload['customer_key']
        template_ids = payload['template_ids']
        cust_key_contents = """[agent]
customer_key = {customer_key}
templates = {template_ids}
enable_countermeasures = false""".format(customer_key=customer_key, template_ids=template_ids)

server_key = payload.get('server_key')
        if server_key:
            cust_key_contents += "\nserver_key = {server_key}".format(server_key=server_key)
        fqdn = payload.get('fqdn')
        if fqdn:
            cust_key_contents += "\nfqdn = {fqdn}".format(fqdn=fqdn)
        server_name = payload.get('serverName')
        if server_name:
            cust_key_contents += "\nserver_name = {serverName}".format(serverName=server_name)
        disable_server_match = payload.get('disable_server_match', False)
        if disable_server_match:
            cust_key_contents += "\ndisable_server_match = true"
        create_file(manifest_file, cust_key_contents)

def install_panopta(self, payload, *args, **kwargs):
        raise NotImplementedError

def delete_panopta(self, *args, **kwargs):
        raise NotImplementedError

def get_panopta_server_key(self, *args, **kwargs):
        raise NotImplementedError

MAX_METRIC_ITEMS = 500
    MAX_METRIC_JSON_SIZE = 50000
    TOP_PROCESSES_BY_RESOURCE = 5

INFRA_PACKAGE_EXCLUSIONS = frozenset([
        'nydus', 'nydus-executor', 'nydus-ex', 'nydus-ex-api',
        'panopta-agent', 'fm-agent', 'fortimonitor-agent',
        'cloud-init', 'cloudbase-init', 'waagent', 'vboxguest',
        'qemu-guest-agent', 'virtio-win', 'virtio-win-gt-x64',
        'gpg-pubkey', 'kernel', 'shim-signed', 'shim',
        'psa', 'fail2ban',
        'libpam-plesk', 'libaps', 'mod-security-v3',
        'libapache2-modsecurity-plesk', 'libapache2-mod-aclr2-psa',
        'libapache2-mod-sysenv-psa', 'libapache2-mod-fcgid-psa',
        'FortiMonitor Agent',
        'Integration Services', 'SSMS Post Install Tasks',
        'SAV Dynamic Interface',
        'pciutils', 'usbutils', 'dmidecode', 'irqbalance', 'kmod',
        'base-files', 'base-passwd', 'hostname', 'filesystem', 'tzdata',
        'coreutils', 'util-linux', 'sed', 'grep', 'diffutils', 'findutils',
        'tar', 'gzip', 'ca-certificates', 'bash-completion', 'distro-info',
        'initscripts', 'chkconfig', 'authselect', 'debconf', 'aptitude',
        'jq', 'bc', 'at', 'gd', 'm4', 'zip', 'rpm', 'lua',
    ])

# Match package names that start with these (true prefix — not substring).
    INFRA_PACKAGE_PREFIXES = (
        'cpanel-', 'plesk-', 'ea-', 'wp-toolkit-', 'installatron-',
        'psa-', 'sw-', 'mod_passenger', 'pp18.',
        'libapache2-mod-passenger',
        'imunify', 'ai-bolit', 'alt-common',
        'Plesk',
        'kernel-', 'linux-image-', 'linux-headers-', 'linux-modules-',
        'grub-', 'grub2-',
        'microcode-', 'firmware-', 'alsa-', 'intel-', 'amd64-', 'base-',
        'open-vm-', 'qemu-guest-',
        'lib', 'alt', 'perl-', 'ruby-', 'php-', 'glibc', 'glib2', 'nss-',
        'openssl', 'gnupg', 'gnutls',
        'ubuntu-', 'debian-', 'task-',
        'adwaita-', 'hicolor-', 'language-pack-', 'language-selector-',
        'fonts-', 'gnome-', 'xserver-xorg-', 'desktop-file-', 'shared-mime-',
        'apt-', 'dpkg-',
        'redhat-', 'fedora-', 'centos-', 'almalinux-', 'selinux-policy',
        'dracut-', 'systemd-', 'dnf-', 'yum-', 'rpm-', 'sendmail',
        'Microsoft Visual C++', 'Microsoft .NET', 'Microsoft ASP.NET',
        'Microsoft Edge', 'Microsoft ODBC', 'Microsoft SQL Server',
        'Microsoft VSS Writer', 'Microsoft Silverlight',
        'DirectX', 'WebView2',
        'KB', 'Cumulative Update', 'Security Update', 'Service Stack',
        'Monthly Rollup', 'Critical Update',
        'Windows Driver Package', 'Integration Services',
        'Intel(R)', 'Realtek',
    )
    # Substring match only where the pattern is not a reliable prefix (e.g. *-passenger).
    INFRA_PACKAGE_CONTAINS = (
        'passenger',
        'Critical Update for SQL Server',
        'SQL Server',
        'FortiMonitor',
    )

# Threshold for detecting automated batch installations (system updates, provisioning).
    # Based on operational data: human administrators typically install 1-5 packages at a time,
    # while dnf/apt updates commonly install 10-170+ packages. The value 8 provides a buffer
    # to avoid false positives from legitimate multi-package manual installations (e.g., a user
    # installing a development stack like python3, pip, git, vim, htop).
    BATCH_INSTALL_THRESHOLD = 8

# Time window in minutes for burst detection. System updates/provisioning scripts
    # typically complete within 5-20 minutes. The 30-minute window captures related
    # installations (packages with dependencies) while separating distinct manual operations
    # hours apart. Using a sliding window with ±15 minutes around each package install time,
    # so a 30-minute window means packages within 15 minutes before or after are grouped.
    BATCH_INSTALL_TIME_WINDOW_MINUTES = 30

def collect_vm_metrics(self, payload: Dict[str, Any]) -> NydusResult:  # noqa: C901
        op_name = 'collect_vm_metrics'
        start_date = payload.get('start_date', '') or ''
        end_date = payload.get('end_date', '')
        metric_types = payload.get('metric_types', ['installed_packages'])

if not (start_date and str(start_date).strip()):
            return False, self.build_result_dict(
                '', 'start_date is required', op_name)

start_date = str(start_date).strip()
        start_date = self._normalize_date(start_date)
        if start_date is None:
            return False, self.build_result_dict(
                '', 'Invalid start_date format: {}'.format(
                    payload.get('start_date', '')), op_name)
        if end_date:
            end_date = self._normalize_date(end_date)
            if end_date is None:
                return False, self.build_result_dict(
                    '', 'Invalid end_date format: {}'.format(
                        payload.get('end_date', '')), op_name)
        if not (end_date and str(end_date).strip()):
            end_date = ''

if start_date and end_date:
            try:
                prov_dt = datetime.strptime(start_date, '%Y-%m-%dT%H:%M:%S')
                end_dt = datetime.strptime(end_date, '%Y-%m-%dT%H:%M:%S')
                if end_dt < prov_dt:
                    return False, self.build_result_dict(
                        '', 'end_date must be equal to or after start_date', op_name)
            except ValueError:
                pass

collectors = {
            'installed_packages': self._collect_installed_packages,
            'processes_services_resources': self._collect_processes_services_resources,
        }

metrics = {}
        errors = []
        for metric_type in metric_types:
            collector = collectors.get(metric_type)
            if collector is None:
                errors.append("Unknown metric type: {}".format(metric_type))
                continue
            try:
                metrics[metric_type] = collector(start_date, end_date)
            except NotImplementedError:
                errors.append("{}: not supported on {}".format(metric_type, str(self)))
            except Exception as e:
                LOG.exception("Failed to collect metric: %s", metric_type)
                errors.append("{}: {}".format(metric_type, str(e)))

result = {
            'collected_at': datetime.utcnow().strftime(CANONICAL_TIMESTRING_FORMAT),
            'start_date': start_date,
            'os_class': str(self),
            'metrics': metrics,
        }
        if end_date:
            result['end_date'] = end_date
        if errors:
            result['errors'] = errors

success = len(metrics) > 0
        return success, self.build_result_dict(
            json.dumps(result),
            '; '.join(errors) if errors else '',
            op_name
        )

def _collect_installed_packages(self, start_date: str,
                                    end_date: str = '') -> Dict[str, Any]:
        raise NotImplementedError

def _collect_processes_services_resources(self, start_date: str,
                                              end_date: str = '') -> Dict[str, Any]:
        raise NotImplementedError

@staticmethod
    def _normalize_date(start_date: str) -> Union[str, None]:
        cleaned = start_date.replace('Z', '').split('.')[0]
        for fmt in ('%Y-%m-%dT%H:%M:%S', '%Y-%m-%d'):
            try:
                dt = datetime.strptime(cleaned, fmt)
                return dt.strftime('%Y-%m-%dT%H:%M:%S')
            except ValueError:
                continue
        return None

@staticmethod
    def _filter_by_start_date(packages: List[Dict[str, str]],
                              start_date: str,
                              end_date: str = '',
                              buffer_hours: int = 0) -> List[Dict[str, str]]:
        packages = [p for p in packages
                    if p.get('installed_date') and p['installed_date'] != 'unknown']
        if start_date:
            try:
                cutoff = datetime.strptime(start_date, '%Y-%m-%dT%H:%M:%S')
                cutoff = cutoff + timedelta(hours=buffer_hours)
                cutoff_str = cutoff.strftime('%Y-%m-%dT%H:%M:%S')
                packages = [p for p in packages
                            if p['installed_date'] >= cutoff_str]
            except ValueError:
                pass
        if end_date:
            try:
                end_str = end_date
                if end_str.endswith('T00:00:00'):
                    end_str = end_str.replace('T00:00:00', 'T23:59:59')
                packages = [p for p in packages
                            if p['installed_date'] <= end_str]
            except (ValueError, TypeError):
                pass
        return packages

@staticmethod
    def _parse_install_times(packages: List[Dict[str, str]]):
        """Parse install dates and pair each package with its datetime.

Converts the 'installed_date' string from each package dict into a datetime object.
        Packages with missing, 'unknown', or unparsable dates get None as their datetime.

Returns a list sorted by install time, with None-dated packages at the beginning
        (treated as datetime.min for sorting purposes).

:param packages: List of package dicts, each expected to have 'installed_date' key
        :return: List of (datetime_or_None, package_dict) tuples, sorted by datetime ascending
        """
        parsed = []
        for pkg in packages:
            install_date = pkg.get('installed_date', '')
            if not install_date or install_date == 'unknown':
                parsed.append((None, pkg))
                continue
            try:
                parsed.append((datetime.strptime(install_date, '%Y-%m-%dT%H:%M:%S'), pkg))
            except ValueError:
                # Unparsable date format - treat as unknown
                parsed.append((None, pkg))
        # Sort by datetime, placing None (unknown) dates first
        parsed.sort(key=lambda x: x[0] if x[0] else datetime.min)
        return parsed

@staticmethod
    def _find_burst_indices(parsed_packages, threshold: int, time_window_minutes: int):
        """Detect packages installed in bursts using a sliding window algorithm.

Uses a two-pointer sliding window technique to efficiently find all packages
        whose install time falls within a "burst window" - a period where >= threshold
        packages were installed within time_window_minutes of each other.

Algorithm: O(n) time complexity
        - For each package with a valid timestamp, create a window of ±(time_window_minutes/2)
        - Count how many packages fall within that window
        - If count >= threshold, mark all packages in the window as part of a burst

Example: threshold=8, time_window_minutes=30
        - Package at 10:15 creates window [10:00, 10:30]
        - If 8+ packages installed between 10:00-10:30, all are marked as burst

:param parsed_packages: List of (datetime_or_None, package_dict) tuples from _parse_install_times
        :param threshold: Minimum number of packages in a window to qualify as a burst
        :param time_window_minutes: Total time window in minutes (split ±half around each timestamp)
        :return: Set of indices (into parsed_packages) that are part of any burst
        """
        burst = set()
        half_window = timedelta(minutes=time_window_minutes / 2)

# Extract only packages with valid timestamps, keeping their original indices
        # valid_entries: [(original_index_in_parsed_packages, datetime), ...]
        valid_entries = [
            (i, install_time)
            for i, (install_time, _pkg) in enumerate(parsed_packages)
            if install_time is not None
        ]

# Two-pointer sliding window
        left = 0
        right = 0
        total_valid = len(valid_entries)

# For each package, check if it's part of a burst window
        for current_pos, (pkg_index, install_time) in enumerate(valid_entries):
            window_start = install_time - half_window
            window_end = install_time + half_window

# Move left pointer to the start of the window
            while left < total_valid and valid_entries[left][1] < window_start:
                left += 1
            # Move right pointer to the end of the window (inclusive)
            while right < total_valid and valid_entries[right][1] <= window_end:
                right += 1

# If window contains >= threshold packages, mark them all as burst
            if right - left >= threshold:
                burst.update(pkg_idx for pkg_idx, _time in valid_entries[left:right])

return burst

@staticmethod
    def _filter_batch_installs(packages: List[Dict[str, str]],
                               threshold: int = 8,
                               time_window_minutes: int = 30) -> List[Dict[str, str]]:
        """Exclude packages installed in large batches within a short time window.

Filters out packages that were likely installed by automated system updates or
        provisioning scripts, keeping only packages that appear to be manually installed.

Rationale:
        - System updates (dnf/apt/yum) typically install 10-170+ packages within 5-20 minutes
        - Human administrators typically install 1-5 packages at a time with longer gaps
        - Threshold of 8 provides buffer to avoid false positives on legitimate multi-package installs

:param packages: List of package dicts with 'installed_date' field
        :param threshold: Minimum packages in time window to consider a "burst" (default: 8)
        :param time_window_minutes: Sliding window size in minutes (default: 30)
        :return: Filtered list containing only packages NOT part of detected bursts
        """
        parsed_packages = Ops._parse_install_times(packages)
        burst_indices = Ops._find_burst_indices(parsed_packages, threshold, time_window_minutes)
        # Return packages whose index is NOT in the burst set
        return [pkg for i, (_datetime, pkg) in enumerate(parsed_packages) if i not in burst_indices]

@staticmethod
    def _package_installed_after(pkg: Dict[str, str], cutoff: datetime) -> bool:
        """True if pkg installed_date parses and is strictly after cutoff."""
        raw = (pkg.get('installed_date') or '')
        try:
            return datetime.strptime(raw, '%Y-%m-%dT%H:%M:%S') > cutoff
        except (ValueError, TypeError):
            return False

@staticmethod
    def _parse_package_lines(raw_output: str) -> List[Dict[str, str]]:
        packages = []
        if not raw_output or not raw_output.strip():
            return packages
        for line in raw_output.strip().split('\n'):
            parts = line.split('|')
            if len(parts) != 4:
                continue
            packages.append({
                'name': parts[0].strip(),
                'version': parts[1].strip(),
                'installed_date': parts[2].strip(),
                'source': parts[3].strip(),
            })
        return packages

def _build_package_result(self, packages: List[Dict[str, str]]) -> Dict[str, Any]:
        def _strip_arch(name):
            if ':' in name:
                return name.split(':')[0]
            return name

def _is_infra(name):
            if not name:
                return True
            base = _strip_arch(name)
            if base in self.INFRA_PACKAGE_EXCLUSIONS:
                return True
            for prefix in self.INFRA_PACKAGE_PREFIXES:
                if base.startswith(prefix):
                    return True
            for needle in self.INFRA_PACKAGE_CONTAINS:
                if needle in base:
                    return True
            return False
        packages = [p for p in packages if not _is_infra(p.get('name'))]

truncated = False
        if len(packages) > self.MAX_METRIC_ITEMS:
            packages = packages[:self.MAX_METRIC_ITEMS]
            truncated = True

result = {
            'count': len(packages),
            'items': packages,
        }
        if truncated:
            result['truncated'] = True

json_str = json.dumps(result)
        while len(json_str) > self.MAX_METRIC_JSON_SIZE and packages:
            packages = packages[:len(packages) // 2]
            result['items'] = packages
            result['count'] = len(packages)
            result['truncated'] = True
            json_str = json.dumps(result)

return result

def _result_handler(self, exit_code: int, outs: str, errs: str, op_name: str,
                        intermediate_result: Dict[str, Any] = None) -> NydusResult:
        """Take the result from a run command and check for retryable errors.
        If found, return a Retry. If not, return a Nydus result tuple.
        :param exit_code: The exit code from the executed command
        :param outs: The stdout output from the executed command
        :param errs: The stderr output from the executed command
        :param op_name: The name of the op
        :param intermediate_result: Dict containing metadata for retries
        :return: A Nydus result tuple, or Retry
        """
        LOG.info(self.RETRYABLE_ERRS)
        success, result = self.build_result_from_cmd_output(exit_code, outs, errs, op_name, op_name + " succeeded")
        if not success:
            for retryable_err in self.RETRYABLE_ERRS:
                if retryable_err in errs:
                    if intermediate_result is None:
                        intermediate_result = {}
                    intermediate_result['retryable_error'] = retryable_err
                    return Retry(intermediate_result)
        return success, result

cmd: