Command manager development

[havidarou]

Description

The Command Manager accepts command requests from Server's Engine and Management API as well as Indexer's Content manager.

These command requests are expanded into orders and written to the orders stream. Orders have information such as:

Field	Type	Description
agent.groups	keyword	List of groups the agent belong to.
command.source	keyword	Origin of the request. One of [Users/Services (via Management API), Engine (via Management API), Content manager (directly)].
command.user	keyword	The user that originated the request. This user may represent a Management API or Indexer API user depending on the source.
command.target.id	keyword	Unique identifier of the destination to send the command to.
command.target.type	keyword	The destination type. One of [group, agent, server],
command.action.name	keyword	The requested action type. Examples: restart, update, change_group, apply_policy, ...
command.action.args	Object	Array of command arguments, starting with the absolute path to the executable.
command.action.version	keyword	Version of the command's schema.
command.timeout	short	Time window in which the command has to be sent to its target.
command.status	keyword	Status within the Command Manager's context. One of [pending, sent, success, failure].
command.result.code	short	Status code returned by the target.
command.result.message	keyword	Result message returned by the target.
command.result.data	keyword	Result data returned by the target.
command.request_id	keyword	UUID generated by the Command Manager.
command.order_id	keyword	UUID generated by the Command Manager.

Roughly speaking, the Command Manager is composed of two main parts:

• API endpoints used to accept command requests and write the expanded orders in the orders stream.
• A scheduled job is used to process the orders stream, looking for expired orders and managing them (sets its status to "Failure").

For Agents to poll for orders, they must use the Comms API's /poll_commands endpoint. Agents must maintain this polling at all times by sending the /poll_commands request in case it drops.

The orders stream must be maintained. Only relevant orders should be preserved, while the rest should be cleaned/archived.

For example:

1. The Management API sends a command requests to the Command Manager's API to update agents in the Test group.
2. The Command Manager API expands the command request and generates an order for each agent in the Test group. These orders are written into the orders stream.
3. The Management API queries the orders stream for pending orders and distributes them through the Server, so that the update order is sent to every agent via the Comms API's /poll_commands endpoint.
4. After every agent in the Test groups has been updated, they report back to the Comms API with the result.
5. The Comms API updates the order status in the orders stream based on the Agent's response.

flowchart TD

subgraph Agents
    Endpoints
    Clouds
    Other_sources
end

subgraph Indexer["Indexer cluster"]
    subgraph Data_states["Data streams"]
        commands_stream["Orders stream"]
    end

    subgraph indexer_modules["Indexer modules"]
        commands_manager["Commands manager"]
        content_manager["Content manager"]
    end
end

subgraph Wazuh1["Server 1"]
    comms_api["Comms API"]
    engine["Engine"]
    management_api["Management API"]
    server["Server"]
end

subgraph Dashboard
    subgraph Dashboard1["Dashboard"]
    end
end

subgraph lb["Load Balancer"]
    lb_node["Per request"]
end


Agents -- 3.a) /poll_commands --> lb
lb -- 3.a) /poll_commands --> comms_api

content_manager -- 1.a) /send_commands --> commands_manager
management_api -- 1.a) /send_commands --> commands_manager
commands_manager -- 1.b) /index --> commands_stream

server -- 2.a) /get_commands --> commands_stream
server -- 2.b) /send_commands --> comms_api
server -- 2.b) /send_commands --> engine

users["Wazuh users"] --> Dashboard
Dashboard -- HTTP --> Indexer

style Data_states fill:#abc2eb
style indexer_modules fill:#abc2eb

Loading

---
title: Command Manager context diagram
---
graph TD
    subgraph Command_Manager["Command Manager"]
        API["Commands API"]
        Controller["Commands Controller"]
        Processor["Commands Expansion"]
        Storage["Commands Index Storage"]
        CommandsIndex[(commands index)]
        AgentsIndex[(agents index)]
        Scheduler["Job Scheduler Task"]
    end

    Actor("Actor") -- POST /commands --> API

    API --> Controller
    Controller --> Processor
    Processor --> Storage
    Storage -- write --> CommandsIndex
    Processor -- read --> AgentsIndex
    Scheduler -- read-write--> CommandsIndex

    subgraph Server["Server"]
        direction TB
        ManagementAPI["Management API"]
    end

    ManagementAPI -- read --> CommandsIndex

Loading

Functional requirements

• The Command Manager exposes an API to accept command requests.
• The Command Manager expands command requests into orders.
• Command requests can target agents, agent groups and server.
• The generated orders are written the orders stream.
• The Command Manager must regularly process the orders stream for its maintainance.

Plan

Spike

• Command manager plugin's scaffolding wazuh-indexer-plugins#56
  • Owner: @wazuh/devel-indexer
• Research how to implement a plugin that extends the job-scheduler wazuh-indexer-plugins#65
  • Owner: @wazuh/devel-indexer
• Build and include the Command manager plugin to the wazuh-indexer packages #407
  • Owner: @wazuh/devel-indexer
• Research how to send commands back to the Wazuh Server  wazuh-indexer-plugins#73
  • Owner: @wazuh/devel-indexer
• Research configuration persistence methods wazuh-indexer-plugins#86
  • Owner: @wazuh/devel-indexer

MVP

• Implement commands index creation wazuh-indexer-plugins#42
  • Owner: @wazuh/devel-indexer
• Implement the Rest API on the Indexer wazuh-indexer-plugins#69
  • Owner: @wazuh/devel-indexer
• Implement HTTP client logic wazuh-indexer-plugins#89
  • Owner: @wazuh/devel-indexer
• Implement the configuration persistence logic wazuh-indexer-plugins#95
  • Owner: @wazuh/devel-indexer
• Implement the job-scheduler logic wazuh-indexer-plugins#87
  • Owner: @wazuh/devel-indexer
• Implement commands sending to Indexer wazuh#26077
  • Owner: @wazuh/devel-pyserver
• Implement server orders distribution wazuh#25599
  • Owner: @wazuh/devel-pyserver

Checkpoint

• Review the Comms API POST /events/stateful endpoint when receiving commands results wazuh#25835
  • Owner: @wazuh/devel-pyserver
• MVP Data Persistence checkpoint #588

MVP redesign

• Adapt Command Manager to the new design wazuh-indexer-plugins#203
• Improve interoperability of commands by changing action.args to object #618
• Command Manager configuration file wazuh-indexer-plugins#180
• Make agents and commands indices visible wazuh-indexer-plugins#240
• Error on commands manager arguments parsing wazuh-indexer-plugins#244
• [BUG] Infinite loop during the pagination of search results wazuh-indexer-plugins#226
• New commands maintenance design wazuh-indexer-plugins#249
• Dedicated command.action.args models wazuh-indexer-plugins#248

[AlexRuiz7]

Blocked due to incoming changes on the design.

These sub-issues move to on hold too.

• Command Manager configuration file wazuh-indexer-plugins#180
• [BUG] The Command Manager sends the same command more than once wazuh-indexer-plugins#195
• [BUG] testPost_success is flaky wazuh-indexer-plugins#163
• [BUG] The Command Manager re-autenticates on each request wazuh-indexer-plugins#192
• Add Management API CA file field to opensearch.yml wazuh-indexer-plugins#171