Research how to send commands back to the Wazuh Server

[AlexRuiz7]

Description

The Command Manager will send the processed commands back to the Wazuh Server for their delivery to the final target.

We need to explore how to send this information to the Management API on the Wazuh Server (outbound HTTP traffic) from the Command Manager. If possible, we should investigate OpenSearch has this kind of functionality of its core or any of its plugins, and reuse any Java dependency already part of it.

[f-galland]

After internal discussion, it was determined we are to use the Apache HttpCore library to perform outgoing HTTP requests.

I've built a PoC plugin with a class following the examples from apache's documentation.

The PoC mostly boils down to the following class (at the moment of writing):

AsyncClientHttpExchange

/*
 * ====================================================================
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * ====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of the Apache Software Foundation.  For more
 * information on the Apache Software Foundation, please see
 * <http://www.apache.org/>.
 *
 */
package org.opensearch.path.to.plugin;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.http.HttpConnection;
import org.apache.hc.core5.http.HttpHost;
import org.apache.hc.core5.http.HttpRequest;
import org.apache.hc.core5.http.HttpResponse;
import org.apache.hc.core5.http.Message;
import org.apache.hc.core5.io.CloseMode;
import org.apache.hc.core5.http.impl.Http1StreamListener;
import org.apache.hc.core5.http.impl.bootstrap.AsyncRequesterBootstrap;
import org.apache.hc.core5.http.impl.bootstrap.HttpAsyncRequester;
import org.apache.hc.core5.http.message.RequestLine;
import org.apache.hc.core5.http.message.StatusLine;
import org.apache.hc.core5.http.nio.entity.StringAsyncEntityConsumer;
import org.apache.hc.core5.http.nio.support.AsyncRequestBuilder;
import org.apache.hc.core5.http.nio.support.BasicResponseConsumer;
import org.apache.hc.core5.reactor.IOReactorConfig;
import org.apache.hc.core5.util.Timeout;

/**
 * Example of asynchronous HTTP/1.1 request execution.
 */


public class AsyncClientHttpExchange {
    HttpAsyncRequester requester;

    public void prepareAsyncRequest() {
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
            .setSoTimeout(5, TimeUnit.SECONDS)
            .build();

        // Create and start requester
        this.requester = AsyncRequesterBootstrap.bootstrap()
            .setIOReactorConfig(ioReactorConfig)
            .setStreamListener(new Http1StreamListener() {

                @Override
                public void onRequestHead(final HttpConnection connection, final HttpRequest request) {
                    System.out.println(connection.getRemoteAddress() + " " + new RequestLine(request));
                }

                @Override
                public void onResponseHead(final HttpConnection connection, final HttpResponse response) {
                    System.out.println(connection.getRemoteAddress() + " " + new StatusLine(response));
                }

                @Override
                public void onExchangeComplete(final HttpConnection connection, final boolean keepAlive) {
                    if (keepAlive) {
                        System.out.println(connection.getRemoteAddress() + " exchange completed (connection kept alive)");
                    } else {
                        System.out.println(connection.getRemoteAddress() + " exchange completed (connection closed)");
                    }
                }

            })
            .create();
    }

    public CompletableFuture<String> performAsyncRequest() throws Exception {
        this.requester.start();
        final HttpHost target = OutgoingHTTPRequestPlugin.TARGET;
        final String requestUri = OutgoingHTTPRequestPlugin.REQUEST_URI;

        CompletableFuture<String> future = new CompletableFuture<>();

        // @Todo: Modify the plugin to handle multiple requests concurrently

        final CountDownLatch latch = new CountDownLatch(1);
        this.requester.execute(
            AsyncRequestBuilder.get()
                .setHttpHost(target)
                .setPath(requestUri)
                .build(),
            new BasicResponseConsumer<>(new StringAsyncEntityConsumer()),
            Timeout.ofSeconds(5),
            new FutureCallback<Message<HttpResponse, String>>() {

                @Override
                public void completed(final Message<HttpResponse, String> message) {
                    final HttpResponse response = message.getHead();
                    final String body = message.getBody();
                    System.out.println(requestUri + "->" + response.getCode());
                    System.out.println(body);
                    latch.countDown();
                    future.complete(body);
                }

                @Override
                public void failed(final Exception ex) {
                    System.out.println(requestUri + "->" + ex);
                    latch.countDown();
                }

                @Override
                public void cancelled() {
                    System.out.println(requestUri + " cancelled");
                    latch.countDown();
                }

            });

        latch.await();
        System.out.println("Shutting down I/O reactor");
        this.requester.initiateShutdown();
        return future;
    }

    public void close() {
        System.out.println("HTTP requester shutting down");
        this.requester.close(CloseMode.GRACEFUL);
    }
}

The prepareAsyncRequest() and performAsyncRequest() methods get called in the following manner:

    @Override
    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
        asyncClientHttpExchange = new AsyncClientHttpExchange();

        return channel -> {
            this.asyncClientHttpExchange.prepareAsyncRequest();
            this.asyncClientHttpExchange.performAsyncRequest()
                .thenAccept(
                    response -> {
                        channel.sendResponse(new BytesRestResponse(RestStatus.OK, response));
                    }
                ).exceptionally(
                    e -> {
                        channel.sendResponse(new BytesRestResponse(RestStatus.INTERNAL_SERVER_ERROR, e.getMessage()));
                        return null;
                    }
                );
        };

    }

The intent of this is for the plugin to respond to a GET method to its only endpoint by issuing an asynchronous HTTP request to a third party REST API, retrieve the json contents of the reply as a String and then make them available as the reply's body (showing it in a curl command's console output).

Though the code is rough on the edges, the plugin does get to be loaded and run, but when issuing a request, it throws a socket permissions error:

[2024-10-01T20:24:03,467][INFO ][stdout                   ] [integTest-0] /todos/1->java.security.AccessControlException: access denied ("java.net.SocketPermission" "172.67.167.151:80" "connect,resolve")

A forum post from opensearch suggests creating a plugin-security.policy file under src/main/plugin-metadata with the following contents:

grant {
    permission java.net.SocketPermission “*:80”, “connect,resolve”;
};

I have tried this with no luck so far.

I'll continue to investigate the issue...

[f-galland]

It looks like the issue lied in the double quotes symbol used in the plugin-security.policy file. I changed them to regular double quotes and the plugin worked:

grant {
permission java.net.SocketPermission "*:80", "connect,resolve";
};

$ curl http://localhost:9200/_plugins/http-requests
{
  "userId": 1,
  "id": 1,
  "title": "delectus aut autem",
  "completed": false
}fede@tyner:~/src/opensearch-plugin-template-java (main *+)
$ 

[AlexRuiz7]

Understood. The approach looks great, although the chosen example seems to be a bit messy and over-complicated for our current use case.

While investigating the Apache HTTP library, I stumped upon this other example that achieves the same goal, executing an asynchronous HTTP request. Although the code is quite similar, it's much shorter, direct and simple to read and understand.

Comparing the examples, I noticed that the example referred here comes from the Apache HttpCore library, while the one posted in this comment comes from Apache HttpClient library. I assume the differences between both is that HttpCore contains the building blocks of the library while the HttpClient one already places some of these pieces in place, so using Apache's Http library is easier and more straightforward to use. That assumption is confirmed here.

I therefore think that we should go for the HttpClient example as the base building block for our Http service, instead of HttpCore's example, which adds advanced mechanism such as thread's synchronization (CountDownLatch) which we don't need for now.

A version of the example extended with a POST HTTP request is attached:

    implementation("org.apache.httpcomponents.client5:httpclient5:5.4")

/*
 * ====================================================================
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * ====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of the Apache Software Foundation.  For more
 * information on the Apache Software Foundation, please see
 * <http://www.apache.org/>.
 *
 */
package org.example;

import java.util.concurrent.Future;

import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
import org.apache.hc.client5.http.async.methods.SimpleHttpResponse;
import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder;
import org.apache.hc.client5.http.async.methods.SimpleRequestProducer;
import org.apache.hc.client5.http.async.methods.SimpleResponseConsumer;
import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.HttpHost;
import org.apache.hc.core5.http.message.StatusLine;
import org.apache.hc.core5.io.CloseMode;
import org.apache.hc.core5.reactor.IOReactorConfig;
import org.apache.hc.core5.util.Timeout;

/**
 * Example of asynchronous HTTP/1.1 request execution.
 */
public class AsyncClientHttpExchange {

    public static void main(final String[] args) throws Exception {

        final IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setSoTimeout(Timeout.ofSeconds(5))
                .build();

        final CloseableHttpAsyncClient client = HttpAsyncClients.custom()
                .setIOReactorConfig(ioReactorConfig)
                .build();

        client.start();

        final HttpHost target = new HttpHost("http", "localhost", 5000);
        final String[] requestUris = new String[] {"/orders"};

        for (final String requestUri: requestUris) {
            // GET
            final SimpleHttpRequest getRequest = SimpleRequestBuilder.get()
                    .setHttpHost(target)
                    .setPath(requestUri)
                    .build();

            System.out.println("Executing request " + getRequest);
            final Future<SimpleHttpResponse> future = client.execute(
                    SimpleRequestProducer.create(getRequest),
                    SimpleResponseConsumer.create(),
                    getCallback(getRequest));
            future.get();

            // POST
            String message = "{\"message\" : \"Hello world!\" }";
            final SimpleHttpRequest postRequest = SimpleRequestBuilder.post()
                    .setHttpHost(target)
                    .setPath(requestUri)
                    .setBody(message, ContentType.APPLICATION_JSON)
                    .build();

            System.out.println("Executing request " + postRequest);
            final Future<SimpleHttpResponse> future2 = client.execute(
                    SimpleRequestProducer.create(postRequest),
                    SimpleResponseConsumer.create(),
                    getCallback(postRequest));
            future2.get();
        }

        System.out.println("Shutting down");
        client.close(CloseMode.GRACEFUL);
    }

    private static FutureCallback<SimpleHttpResponse> getCallback(SimpleHttpRequest request) {
        return new FutureCallback<SimpleHttpResponse>() {

            @Override
            public void completed(final SimpleHttpResponse response) {
                System.out.println(request + "->" + new StatusLine(response));
                System.out.println(response.getBody().getBodyText());
            }

            @Override
            public void failed(final Exception ex) {
                System.out.println(request + "->" + ex);
            }

            @Override
            public void cancelled() {
                System.out.println(request + " cancelled");
            }

        };
    }

}

Result of this code execution:

Executing request GET http://localhost:5000/orders
Executing request POST http://localhost:5000/orders
Shutting down
GET http://localhost:5000/orders->HTTP/1.1 200 OK
POST http://localhost:5000/orders->HTTP/1.1 201 Created

GET /orders
Headers
---------
Host: localhost:5000
Connection: keep-alive,Upgrade
User-Agent: Apache-HttpAsyncClient/5.4 (Java/21.0.4)
Upgrade: TLS/1.2


127.0.0.1 - - [07/Oct/2024 12:51:20] "GET /orders HTTP/1.1" 200 -

POST /orders
Headers
---------
Host: localhost:5000
Content-Length: 29
Content-Type: application/json; charset=UTF-8
Connection: keep-alive
User-Agent: Apache-HttpAsyncClient/5.4 (Java/21.0.4)


JSON
---------
{
    "message": "Hello world!"
}
127.0.0.1 - - [07/Oct/2024 12:51:20] "POST /orders HTTP/1.1" 201 -