feat: execute shell scripts created via redirect (cowrie#2908)

Author: micheloosterhof

src/cowrie/commands/bash.py

@@ -1,23 +1,26 @@
-# Copyright (c) 2009 Upi Tamminen <desaster@gmail.com>
-# See the COPYRIGHT file for more information
-
-# coding=utf-8
+# ABOUTME: Implements bash/sh shell command for cowrie honeypot.
+# ABOUTME: Handles -c flags, piped input, script file execution, and interactive shells.
 
 from __future__ import annotations
 
+import re
 from typing import TYPE_CHECKING
 
 from twisted.internet import error
 from twisted.python import failure
 
 from cowrie.shell.command import HoneyPotCommand
+from cowrie.shell.fs import FileNotFound
 from cowrie.shell.honeypot import HoneyPotShell
 
 if TYPE_CHECKING:
     from collections.abc import Callable
 
 commands: dict[str, Callable] = {}
 
+MAX_SCRIPT_DEPTH = 5
+_SHEBANG_RE = re.compile(r"^#!\s*/bin/(ba)?sh")
+
 
 class Command_sh(HoneyPotCommand):
     def start(self) -> None:
@@ -37,10 +40,41 @@ def start(self) -> None:
             self.execute_commands(self.input_data.decode("utf8"))
             self.exit()
 
+        elif self.args and not self.args[0].startswith("-"):
+            self.execute_script_file(self.args[0])
+            self.exit()
+
         else:
             self.interactive_shell()
 
-        # TODO: handle spawning multiple shells, support other sh flags
+    def execute_script_file(self, filename: str) -> None:
+        depth = getattr(self.protocol, "_script_depth", 0)
+        if depth >= MAX_SCRIPT_DEPTH:
+            self.errorWrite(f"-bash: {filename}: too many levels of recursion\n")
+            return
+
+        path = self.fs.resolve_path(filename, self.protocol.cwd)
+        try:
+            contents = self.fs.file_contents(path)
+        except (FileNotFound, FileNotFoundError):
+            self.errorWrite(f"bash: {filename}: No such file or directory\n")
+            return
+
+        lines = contents.decode("utf-8", errors="replace").splitlines()
+        # Strip shebang line
+        if lines and _SHEBANG_RE.match(lines[0]):
+            lines = lines[1:]
+        # Strip comment-only lines and blank lines
+        lines = [line for line in lines if line.strip() and not line.strip().startswith("#")]
+
+        if not lines:
+            return
+
+        self.protocol._script_depth = depth + 1
+        try:
+            self.execute_commands("; ".join(lines))
+        finally:
+            self.protocol._script_depth = depth
 
     def execute_commands(self, cmds: str) -> None:
         # self.input_data holds commands passed via PIPE

src/cowrie/shell/protocol.py

@@ -143,6 +143,69 @@ def call(self):
 
         return Command_txtcmd
 
+    def scriptcmd(self, path: str) -> object:
+        """Return a command class that executes a shell script from the virtual filesystem."""
+        import re
+
+        shebang_re = re.compile(r"^#!\s*/bin/(ba|a)?sh")
+        max_depth = 5
+
+        class Command_scriptcmd(command.HoneyPotCommand):
+            def call(self_cmd):
+                depth = getattr(self_cmd.protocol, "_script_depth", 0)
+                if depth >= max_depth:
+                    self_cmd.errorWrite(
+                        f"-bash: {path}: too many levels of recursion\n"
+                    )
+                    return
+
+                try:
+                    contents = self_cmd.fs.file_contents(path)
+                except Exception:
+                    self_cmd.errorWrite(
+                        f"-bash: {path}: No such file or directory\n"
+                    )
+                    return
+
+                # Null bytes indicate actual binary — reject like real bash
+                if b"\x00" in contents:
+                    self_cmd.errorWrite(
+                        f"-bash: {path}: cannot execute binary file: Exec format error\n"
+                    )
+                    return
+
+                lines = contents.decode("utf-8", errors="replace").splitlines()
+
+                if not lines:
+                    return
+
+                # Strip shebang line if it's a shell shebang
+                if shebang_re.match(lines[0]):
+                    lines = lines[1:]
+
+                # Strip comment-only and blank lines
+                lines = [
+                    line
+                    for line in lines
+                    if line.strip() and not line.strip().startswith("#")
+                ]
+
+                if not lines:
+                    return
+
+                self_cmd.protocol._script_depth = depth + 1
+                try:
+                    shell = honeypot.HoneyPotShell(
+                        self_cmd.protocol, interactive=False
+                    )
+                    self_cmd.protocol.cmdstack.append(shell)
+                    shell.lineReceived("; ".join(lines))
+                    self_cmd.protocol.cmdstack.pop()
+                finally:
+                    self_cmd.protocol._script_depth = depth
+
+        return Command_scriptcmd
+
     def isCommand(self, cmd):
         """
         Check if cmd (the argument of a command) is a command, too.
@@ -187,6 +250,9 @@ def getCommand(self, cmd, paths):
         if path in self.commands:
             return self.commands[path]
 
+        if self.fs.isfile(path):
+            return self.scriptcmd(path)
+
         log.msg(f"Can't find command {cmd}")
         return None
 

src/cowrie/test/test_script_execution.py

@@ -0,0 +1,119 @@
+# ABOUTME: Tests for shell script execution via bash/sh and path-based invocation.
+# ABOUTME: Covers shebang stripping, binary detection, comment stripping, and recursion depth limits.
+
+from __future__ import annotations
+
+import os
+import unittest
+
+from cowrie.shell.protocol import HoneyPotInteractiveProtocol
+from cowrie.test.fake_server import FakeAvatar, FakeServer
+from cowrie.test.fake_transport import FakeTransport
+
+os.environ["COWRIE_HONEYPOT_DATA_PATH"] = "data"
+os.environ["COWRIE_HONEYPOT_DOWNLOAD_PATH"] = "/tmp"
+os.environ["COWRIE_SHELL_FILESYSTEM"] = "src/cowrie/data/fs.pickle"
+
+PROMPT = b"root@unitTest:~# "
+
+
+class ScriptExecutionTests(unittest.TestCase):
+    """Tests for executing shell scripts via bash/sh and ./path."""
+
+    def setUp(self) -> None:
+        self.proto = HoneyPotInteractiveProtocol(FakeAvatar(FakeServer()))
+        self.tr = FakeTransport("", "31337")
+        self.proto.makeConnection(self.tr)
+        self.tr.clear()
+
+    def tearDown(self) -> None:
+        self.proto.connectionLost()
+
+    def test_bash_executes_script_file(self) -> None:
+        """bash script.sh reads and executes file contents."""
+        self.proto.lineReceived(b'echo "echo hello" > /tmp/test.sh')
+        self.tr.clear()
+        self.proto.lineReceived(b"bash /tmp/test.sh")
+        self.assertEqual(self.tr.value(), b"hello\n" + PROMPT)
+
+    def test_sh_executes_script_file(self) -> None:
+        """sh script.sh reads and executes file contents."""
+        self.proto.lineReceived(b'echo "echo world" > /tmp/test_sh.sh')
+        self.tr.clear()
+        self.proto.lineReceived(b"sh /tmp/test_sh.sh")
+        self.assertEqual(self.tr.value(), b"world\n" + PROMPT)
+
+    def test_bash_nonexistent_file(self) -> None:
+        """bash nonexistent.sh shows error."""
+        self.proto.lineReceived(b"bash /tmp/nonexistent.sh")
+        output = self.tr.value()
+        self.assertIn(b"No such file or directory", output)
+
+    def test_dotslash_with_shebang_executes(self) -> None:
+        """./script.sh with #!/bin/sh shebang executes."""
+        self.proto.lineReceived(b'printf "#!/bin/sh\\necho from_script\\n" > run.sh')
+        self.tr.clear()
+        self.proto.lineReceived(b"./run.sh")
+        self.assertEqual(self.tr.value(), b"from_script\n" + PROMPT)
+
+    def test_dotslash_without_shebang_executes(self) -> None:
+        """./file without shebang executes as shell script (kernel ENOEXEC fallback)."""
+        self.proto.lineReceived(b'echo "echo no_shebang" > noshebang.sh')
+        self.tr.clear()
+        self.proto.lineReceived(b"./noshebang.sh")
+        self.assertEqual(self.tr.value(), b"no_shebang\n" + PROMPT)
+
+    def test_dotslash_binary_file_fails(self) -> None:
+        """./file with null bytes emits 'cannot execute binary file'."""
+        # Use printf to write a null byte into the file
+        self.proto.lineReceived(b'printf "\\x00ELF" > binfile')
+        self.tr.clear()
+        self.proto.lineReceived(b"./binfile")
+        output = self.tr.value()
+        self.assertIn(b"cannot execute binary file", output)
+
+    def test_shebang_line_stripped(self) -> None:
+        """Shebang line is not echoed or executed as a command."""
+        self.proto.lineReceived(
+            b'printf "#!/bin/bash\\necho shebang_stripped\\n" > /tmp/shebang.sh'
+        )
+        self.tr.clear()
+        self.proto.lineReceived(b"bash /tmp/shebang.sh")
+        self.assertEqual(self.tr.value(), b"shebang_stripped\n" + PROMPT)
+
+    def test_comment_lines_stripped(self) -> None:
+        """Comment-only lines are stripped from script execution."""
+        self.proto.lineReceived(
+            b'printf "#!/bin/sh\\n# this is a comment\\necho works\\n" > /tmp/comments.sh'
+        )
+        self.tr.clear()
+        self.proto.lineReceived(b"bash /tmp/comments.sh")
+        self.assertEqual(self.tr.value(), b"works\n" + PROMPT)
+
+    def test_multiline_script(self) -> None:
+        """Script with multiple commands executes all of them."""
+        self.proto.lineReceived(
+            b'printf "echo line1\\necho line2\\necho line3\\n" > /tmp/multi.sh'
+        )
+        self.tr.clear()
+        self.proto.lineReceived(b"bash /tmp/multi.sh")
+        self.assertEqual(self.tr.value(), b"line1\nline2\nline3\n" + PROMPT)
+
+    def test_absolute_path_with_shebang(self) -> None:
+        """Absolute path /tmp/script.sh with shebang executes."""
+        self.proto.lineReceived(
+            b'printf "#!/bin/sh\\necho absolute\\n" > /tmp/abs.sh'
+        )
+        self.tr.clear()
+        self.proto.lineReceived(b"/tmp/abs.sh")
+        self.assertEqual(self.tr.value(), b"absolute\n" + PROMPT)
+
+    def test_bash_dash_c_still_works(self) -> None:
+        """Existing sh -c 'cmd' functionality still works."""
+        self.proto.lineReceived(b"sh -c 'echo still_works'")
+        self.assertEqual(self.tr.value(), b"still_works\n" + PROMPT)
+
+    def test_bash_piped_input_still_works(self) -> None:
+        """Existing piped input functionality still works."""
+        self.proto.lineReceived(b"echo echo piped | bash")
+        self.assertEqual(self.tr.value(), b"piped\n" + PROMPT)