# SPDX-FileCopyrightText: 2025 Greenbone AG # # SPDX-License-Identifier: GPL-3.0-or-later from typing import Any, Mapping, Optional, Sequence from gvm.errors import InvalidArgumentType, RequiredArgument from gvm.protocols.core import Request from gvm.protocols.gmp.requests._entity_id import EntityID from gvm.utils import to_bool from gvm.xml import XmlCommand class Agents: @staticmethod def _add_element(element, name: str, value: Any) -> None: """ Helper to add a sub-element with a value if the value is not None. Args: element: The XML parent element to which the new element is added. name: Name of the sub-element to create. value: Value to set as the text of the sub-element. If None, the element will not be created. """ if value is None: return if isinstance(value, bool): value = to_bool(value) else: value = str(value) element.add_element(name, value) @classmethod def _validate_agent_config( cls, config: Mapping[str, Any], *, caller: str ) -> None: """Ensure all required fields exist, are well-shaped, and non-empty.""" def valid_map(d: Any, key: str, path: str) -> Mapping[str, Any]: if not isinstance(d, Mapping): raise RequiredArgument( function=caller, argument=f"config.{path.rstrip('.')}" ) v = d.get(key) if not isinstance(v, Mapping): raise RequiredArgument( function=caller, argument=f"config.{path}{key}" ) return v def valid_value(d: Mapping[str, Any], key: str, path: str) -> Any: v = d.get(key) if v is None or (isinstance(v, str) and v.strip() == ""): raise RequiredArgument( function=caller, argument=f"config.{path}{key}" ) return v # agent_control.retry ac = valid_map(config, "agent_control", "") retry = valid_map(ac, "retry", "agent_control.") valid_value(retry, "attempts", "agent_control.retry.") valid_value(retry, "delay_in_seconds", "agent_control.retry.") valid_value(retry, "max_jitter_in_seconds", "agent_control.retry.") # agent_script_executor se = valid_map(config, "agent_script_executor", "") valid_value(se, "bulk_size", "agent_script_executor.") valid_value(se, "bulk_throttle_time_in_ms", "agent_script_executor.") valid_value(se, "indexer_dir_depth", "agent_script_executor.") sched = se.get("scheduler_cron_time") if isinstance(sched, str): items = [sched] elif isinstance(sched, Sequence) and not isinstance( sched, (str, bytes) ): items = [str(x) for x in sched] else: items = [] if not items or any(not str(x).strip() for x in items): raise RequiredArgument( function=caller, argument="config.agent_script_executor.scheduler_cron_time", ) # heartbeat hb = valid_map(config, "heartbeat", "") valid_value(hb, "interval_in_seconds", "heartbeat.") valid_value(hb, "miss_until_inactive", "heartbeat.") @classmethod def _validate_config_defaults( cls, config_defaults: Mapping[str, Any], *, caller: str ) -> None: """Ensure agent config defaults structure is valid.""" def valid_map(d: Any, key: str, path: str) -> Mapping[str, Any]: if not isinstance(d, Mapping): raise RequiredArgument( function=caller, argument=path.rstrip("."), ) v = d.get(key) if not isinstance(v, Mapping): raise RequiredArgument( function=caller, argument=f"{path}{key}", ) return v def valid_bool(d: Mapping[str, Any], key: str, path: str) -> bool: v = d.get(key) if not isinstance(v, bool): raise InvalidArgumentType( function=caller, argument=f"{path}{key}", arg_type="bool", ) return v agent_defaults = valid_map( config_defaults, "agent_defaults", "config_defaults." ) cls._validate_agent_config(agent_defaults, caller=caller) agent_control_defaults = valid_map( config_defaults, "agent_control_defaults", "config_defaults.", ) valid_bool( agent_control_defaults, "update_to_latest", "config_defaults.agent_control_defaults.", ) @classmethod def _append_agent_config( cls, parent, config: Mapping[str, Any], wrapper_tag: Optional[str] = "config", ) -> None: """ Append an agent configuration block to the given XML parent element. Expected config structure:: { "agent_control": { "retry": { "attempts": 6, "delay_in_seconds": 60, "max_jitter_in_seconds": 10 } }, "agent_script_executor": { "bulk_size": 2, "bulk_throttle_time_in_ms": 300, "indexer_dir_depth": 100, "scheduler_cron_time": ["0 */12 * * *"] }, "heartbeat": { "interval_in_seconds": 300, "miss_until_inactive": 1 } } Args: parent: The XML parent element to which the wrapper element should be appended. config: Mapping containing the agent configuration fields to serialize. wrapper_tag: Optional wrapper element name. If None, fields are appended directly to parent. """ xml_config = ( parent.add_element(wrapper_tag) if wrapper_tag is not None else parent ) # agent_control.retry ac = config["agent_control"] retry = ac["retry"] xml_ac = xml_config.add_element("agent_control") xml_retry = xml_ac.add_element("retry") cls._add_element(xml_retry, "attempts", retry.get("attempts")) cls._add_element( xml_retry, "delay_in_seconds", retry.get("delay_in_seconds") ) cls._add_element( xml_retry, "max_jitter_in_seconds", retry.get("max_jitter_in_seconds"), ) # agent_script_executor se = config["agent_script_executor"] xml_se = xml_config.add_element("agent_script_executor") cls._add_element(xml_se, "bulk_size", se.get("bulk_size")) cls._add_element( xml_se, "bulk_throttle_time_in_ms", se.get("bulk_throttle_time_in_ms"), ) cls._add_element( xml_se, "indexer_dir_depth", se.get("indexer_dir_depth") ) sched = se.get("scheduler_cron_time") if isinstance(sched, str): sched_items = [sched] else: sched_items = list(sched or []) if sched_items: xml_sched = xml_se.add_element( "scheduler_cron_time", attrs={"is_list": "1"}, ) for item in sched_items: xml_sched.add_element("item", str(item)) # heartbeat hb = config["heartbeat"] xml_hb = xml_config.add_element("heartbeat") cls._add_element( xml_hb, "interval_in_seconds", hb.get("interval_in_seconds") ) cls._add_element( xml_hb, "miss_until_inactive", hb.get("miss_until_inactive") ) @classmethod def _append_config_defaults( cls, parent, config_defaults: Mapping[str, Any] ) -> None: xml_defaults = parent.add_element("config_defaults") cls._append_agent_config( xml_defaults, config_defaults["agent_defaults"], wrapper_tag="agent_defaults", ) control_defaults = config_defaults.get("agent_control_defaults") if control_defaults: xml_control_defaults = xml_defaults.add_element( "agent_control_defaults" ) cls._add_element( xml_control_defaults, "update_to_latest", control_defaults.get("update_to_latest"), ) @classmethod def get_agents( cls, *, filter_string: Optional[str] = None, filter_id: Optional[EntityID] = None, details: Optional[bool] = None, ) -> Request: """Request a list of agents. Args: filter_string: Filter term to use for the query. filter_id: UUID of an existing filter to use for the query. details: Whether to include detailed agent info. """ cmd = XmlCommand("get_agents") cmd.add_filter(filter_string, filter_id) if details is not None: cmd.set_attribute("details", to_bool(details)) return cmd @classmethod def modify_agents( cls, agent_ids: list[EntityID], *, authorized: Optional[bool] = None, update_to_latest: Optional[bool] = None, config: Optional[Mapping[str, Any]] = None, comment: Optional[str] = None, ) -> Request: """ Modify multiple agents. Args: agent_ids: List of agent UUIDs to modify. authorized: Whether the agent is authorized. update_to_latest: Whether the agent is allowed to update to latest automatically. config: Nested config, e.g.: { "agent_control": { "retry": { "attempts": 6, "delay_in_seconds": 60, "max_jitter_in_seconds": 10, } }, "agent_script_executor": { "bulk_size": 2, "bulk_throttle_time_in_ms": 300, "indexer_dir_depth": 100, "scheduler_cron_time": ["0 */12 * * *"], # str or list[str] }, "heartbeat": { "interval_in_seconds": 300, "miss_until_inactive": 1, }, } comment: Optional comment for the change. """ if not agent_ids: raise RequiredArgument( function=cls.modify_agents.__name__, argument="agent_ids" ) cmd = XmlCommand("modify_agent") xml_agents = cmd.add_element("agents") for agent_id in agent_ids: xml_agents.add_element("agent", attrs={"id": agent_id}) if authorized is not None: cmd.add_element("authorized", to_bool(authorized)) if update_to_latest is not None: cmd.add_element("update_to_latest", to_bool(update_to_latest)) if config is not None: cls._validate_agent_config( config, caller=cls.modify_agents.__name__ ) cls._append_agent_config(cmd, config) if comment: cmd.add_element("comment", comment) return cmd @classmethod def delete_agents(cls, agent_ids: list[EntityID]) -> Request: """Delete multiple agents Args: agent_ids: List of agent UUIDs to delete """ if not agent_ids: raise RequiredArgument( function=cls.delete_agents.__name__, argument="agent_ids" ) cmd = XmlCommand("delete_agent") xml_agents = cmd.add_element("agents") for agent_id in agent_ids: xml_agents.add_element("agent", attrs={"id": agent_id}) return cmd @classmethod def modify_agent_control_scan_config( cls, agent_control_id: EntityID, config_defaults: Mapping[str, Any], ) -> Request: """ Modify agent control scan config. Args: agent_control_id: The agent control UUID. config_defaults: Nested config, e.g.: { "agent_defaults": { "agent_control": { "retry": { "attempts": 6, "delay_in_seconds": 60, "max_jitter_in_seconds": 10, } }, "agent_script_executor": { "bulk_size": 2, "bulk_throttle_time_in_ms": 300, "indexer_dir_depth": 100, "scheduler_cron_time": ["0 */12 * * *"], }, "heartbeat": { "interval_in_seconds": 300, "miss_until_inactive": 1, }, }, "agent_control_defaults": { "update_to_latest": False, }, } """ if not agent_control_id: raise RequiredArgument( function=cls.modify_agent_control_scan_config.__name__, argument="agent_control_id", ) if not config_defaults: raise RequiredArgument( function=cls.modify_agent_control_scan_config.__name__, argument="config_defaults", ) cls._validate_config_defaults( config_defaults, caller=cls.modify_agent_control_scan_config.__name__, ) cmd = XmlCommand("modify_agent_control_scan_config") cmd.set_attribute("agent_control_id", str(agent_control_id)) cls._append_config_defaults(cmd, config_defaults) return cmd @classmethod def sync_agents(cls) -> Request: """Trigger agents synchronization from all agent controllers.""" return XmlCommand("sync_agents")