Implicit authenticated requests to the Command Manager API

[AlexRuiz7]

Description

In #393, we added authenticated HTTP requests to the Command Manager API. However, this requires the addition of the Indexer's user credentials to the key store, which feels unnecessary or redundant as we are communicating with another puling within the same node.

On this issue, we will explore alternatives to this solution, by investigating how OpenSearch performs requests to the node with implicit authentication (actions such as index creation, deletion and so on).

Functional requirements

• The Content Manager is able to create new commands on the Command Manager successfully.

Implementation restrictions

• The requests to the Command Manager uses implicit authentication.
• Unit tests are provided.
• Integration tests are provided.
• Technical documentation is generated.

Plan

• Explore how OpenSearch performs requests within nodes without explicit authentication.
• Propose possible alternatives to the one current implemented (HTTP auth headers).
• Implement a proof of concept.
• Implement the proof of concept into the Wauzh Indexer plugins.
• Unit and integrations tests.
• Documentation

[QU3B1M]

OpenSearch's Internal Plugin Communication via Transport Layer

OpenSearch provides a mechanism for plugin-to-plugin communication within the same cluster that avoids HTTP overhead. This is done through the Transport Layer using HandledTransportAction. It allows defining internal endpoints that can be invoked via Client#execute() without exposing HTTP routes.

This communication model is ideal for internal plugin logic that doesn’t require REST exposure, improves performance, and leverages OpenSearch’s native action execution flow.

To set up Transport Layer communication between plugins:

1. Define an ActionType to uniquely identify the action.
2. Implement a HandledTransportAction subclass to handle the request/response logic.
3. Register the action in the plugin's getActions() method.
4. Invoke it using Client#execute().

The HandledTransportAction class overrides the doExecute() method, which is triggered when the registered ActionType is executed via the internal client.

---

Proof of Concept: CommandManagerPlugin

• CommandAction: Declares the internal action name and response type

  public class CommandAction extends ActionType<CommandResponse> {
      public static final String NAME = "cluster:command_manager/post_command";
      public static final CommandAction INSTANCE = new CommandAction();
  
      private CommandAction() {
          super(NAME, CommandResponse::new);
      }
  }

• CommandTransport: Implements the internal handler logic

  public class CommandTransport extends HandledTransportAction<CommandRequest, CommandResponse> {
      private static final Logger log = LogManager.getLogger(CommandTransport.class);
      private final Client client;
      private final CommandIndex commandIndex;
  
      @Inject
      public CommandTransport(
          TransportService transportService,
          ActionFilters actionFilters,
          Client client,
          CommandIndex commandIndex
      ) {
          super(CommandAction.NAME, transportService, actionFilters, CommandRequest::new);
          this.client = client;
          this.commandIndex = commandIndex;
      }
  
      @Override
      protected void doExecute(Task task, CommandRequest request, ActionListener<CommandResponse> listener) {
          String jsonBody = request.getJsonBody();
          log.info("Transport Action request received: {}", jsonBody);
  
          try {
              List<Command> commands = Command.parseArray(
                  XContentType.JSON.xContent().createParser(
                      NamedXContentRegistry.EMPTY,
                      DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
                      jsonBody
                  )
              );
  
              if (commands.isEmpty()) {
                  listener.onFailure(new IllegalArgumentException("No commands provided"));
                  return;
              }
  
              Orders orders = Orders.fromCommands(this.client, commands);
              commandIndex.asyncBulkCreate(orders.get())
                  .thenAccept(status -> listener.onResponse(new CommandResponse("Command received: " + jsonBody)))
                  .exceptionally(e -> {
                      listener.onFailure(e instanceof Exception ? (Exception) e : new RuntimeException(e));
                      return null;
                  });
  
          } catch (IOException e) {
              listener.onFailure(e);
          }
      }
  }

• CommandRequest: Encapsulates the request payload

  public class CommandRequest extends ActionRequest {
      private final String jsonBody;
  
      public CommandRequest(String jsonBody) {
          this.jsonBody = jsonBody;
      }
  
      public CommandRequest(StreamInput in) throws IOException {
          super(in);
          this.jsonBody = in.readString();
      }
  
      @Override
      public ActionRequestValidationException validate() {
          return null;
      }
  
      public String getJsonBody() {
          return jsonBody;
      }
  
      @Override
      public void writeTo(StreamOutput out) throws IOException {
          out.writeString(jsonBody);
      }
  }

• CommandResponse: Encapsulates the response

  public class CommandResponse extends ActionResponse {
      private final String message;
  
      public CommandResponse(String message) {
          this.message = message;
      }
  
      public CommandResponse(StreamInput in) throws IOException {
          super(in);
          this.message = in.readString();
      }
  
      public String getMessage() {
          return message;
      }
  
      @Override
      public void writeTo(StreamOutput out) throws IOException {
          out.writeString(message);
      }
  }

• Plugin Registration: CommandManagerPlugin#getActions()

  @Override
  public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
      return List.of(new ActionHandler<>(CommandAction.INSTANCE, CommandTransport.class));
  }

• Executing the Action: Client#execute()

  CommandRequest request = new CommandRequest(requestBody);
  client.execute(
      CommandAction.INSTANCE,
      request,
      new ActionListener<>() {
          @Override
          public void onResponse(CommandResponse response) {
              log.info("Command successfully posted: {}", response.getMessage());
          }
  
          @Override
          public void onFailure(Exception e) {
              log.error("Failed to post command", e);
          }
      }
  );

---

Benefits of Using Transport Layer Communication:

• Efficient: Avoids HTTP stack overhead.
• Secure: Only accessible within the cluster.
• Integrated: Leverages native OpenSearch task/action lifecycle.
• Testable: Easily tested via unit/integration tests without needing REST.

[QU3B1M]

Currently working on the implementation of the Transport's action/request/response on a Command Manager SPI library.

Facing a ClassLoader issue with the shared library, it seems to be because of the method used to implement the library on the plugins, we should replace implementation with compileOnly, but that second method breaks the build process of the Command Manager plugin at the IntegTest step.

Current error when testing the plugins on an actual cluster:

java.lang.ClassCastException: class com.wazuh.commandmanager.spi.CommandRequest cannot be cast to class com.wazuh.commandmanager.spi.CommandRequest (com.wazuh.commandmanager.spi.CommandRequest is in unnamed module of loader java.net.FactoryURLClassLoader @401788d5; com.wazuh.commandmanager.spi.CommandRequest is in unnamed module of loader>
        at com.wazuh.commandmanager.transport.CommandTransportAction.doExecute(CommandTransportAction.java:47) ~[?:?]
        at org.opensearch.action.support.TransportAction$RequestFilterChain.proceed(TransportAction.java:220) [opensearch-2.19.1.jar:2.19.1]
        at org.opensearch.indexmanagement.controlcenter.notification.filter.IndexOperationActionFilter.apply(IndexOperationActionFilter.kt:39) [opensearch-index-management-2.19.1.0.jar:2.19.1.0]