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:

Note

This schema may change.

• Source. One of [Users/Services (via Management API), Engine (via Management API), Content manager (directly)]
• User. The user that originated the request. This user may represent a Management API or Indexer API user depending on the source.
• Target. Cluster name destination.
• Type. One of [Agent groups, Agent, Server cluster]
• Action.
  • Type. One of One of [Agent groups, Agent, Server cluster]
  • Params. Additional parameters for the action.
  • Version. Version of the action.
• Timeout. Number of seconds to wait for the command to be executed.
• Status. One of [Pending, Sent, Success, Failure]
• Result.
  • Code. Result code.
  • Message. Description of the result.
  • Data. Additional data.
• Command Request ID. Unique identifier generated by the Command Manager. Auto-incremental.
• Order ID. Unique identifier generated by the Command Manager. Auto-incremental within the same Command Request ID.
• ID. Document ID. Generated combining the Order ID and the Command Request ID.

The command manager sends every order to its destination for execution by processing the orders stream.

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 scheduler is used to process the orders stream and send every order to its destination for execution.

For Agents to poll for orders, they must use the Comms API's /commands endpoint. Agents must maintain this polling at all times by sending the /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 writes an update order per agent in the Test group in the orders stream.
3. The command manager scheduler processes the orders stream and sends every pending order to its destination for execution via the Management API /accept_commands endpoint.
4. The Management API /accept_commands endpoints distribute orders via the Server's cluster so that the update order is sent to every agent via the Comms API's /commands endpoints.
5. After every agent in the Test groups has been updated, they report back to the Comms API with the result.
6. 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 Server["Server cluster"]

    subgraph cluster["Cluster"]
    end
    subgraph Wazuh1["Server node n"]
        api1["Comms API"]
        engine["Engine"]
        server1["Management </br> API"]

    end

end

subgraph Dashboard
    subgraph Dashboard1["Dashboard"]
    end
end

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

users["Wazuh users"] --> Dashboard
Agents -- /poll_commands --> lb

cluster --> api1
lb -- /poll_commands --> api1
Dashboard -- HTTP --> Indexer
commands_manager -- 1. /index --> commands_stream
commands_manager -- 2. /reads --> commands_stream
content_manager -- /send_commands --> commands_manager
commands_manager -- 3. /respond --> server1
engine -- /send_commands --> commands_manager
server1 -- /send_commands --> commands_manager
server1 -- /distribute_commands --> cluster

cluster --> engine
style Wazuh1 fill:#abc2eb
style Data_states fill:#abc2eb
style Dashboard1 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 Processor"]
        Storage["Commands Index Storage"]
        CommandsIndex[(commands index)]
        AgentsIndex[(agents index)]
        Scheduler["Job Scheduler Task"]
        HTTPClient["HTTP Client"]
        Config["Configuration Service"]
    end

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

    API --> Controller
    Controller --> Processor
    Processor --> Storage
    Storage -- write --> CommandsIndex
    Processor -- read --> AgentsIndex
    CommandsIndex -- read --> Scheduler
    Scheduler --> HTTPClient
    Config -.->|read| HTTPClient

    HTTPClient -- POST /orders --> ManagementAPI["Management API"]

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 the server cluster.
• The command manager expands command requests to orders.
• Orders can only target agents or the server cluster.
• Orders are written the orders stream.
• The command manager must regularly process the orders stream to send pending orders to their destination for execution.
• The command manager sends orders to their destinations for execution via the Server's Management API.

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 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 validation

Feature complete implementation

• Implement the job-scheduler logic wazuh-indexer-plugins#87
  • Owner: @wazuh/devel-indexer
• Implement commands' expansion logic wazuh-indexer-plugins#88
  • Owner: @wazuh/devel-indexer
• Index maintenance: orders cleansing
• RBAC implementation

[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