Fix over SSH (Closes: #234, #137)

CHANGELOG.md

@@ -3,6 +3,10 @@
 Contact: [[email protected]](mailto:[email protected])  
 [https://github.com/ghantoos/lshell](https://github.com/ghantoos/lshell)
 
+
+### v0.10.5 26/10/2024
+- Fixed parsing and testing of the over SSH commands. Corrected return codes over SSH.
+
 ### v0.10.4 26/10/2024
 - Feature: Allow commands with specific parameters, e.g., `telnet localhost`. Adding `telnet localhost` to the `allowed` configuration will not permit the `telnet` command by itself but will only allow the exact match `telnet localhost`.
 

etc/lshell.conf

@@ -24,7 +24,7 @@ loglevel        : 2
 ##  string will disable LD_PRELOAD prepend of the commands. This is done at your
 ##  own risk, as lshell becomes easily breached using some commands like find(1)
 ##  using the -exec flag.
-#path_noexec     : /usr/libexec/sudo_noexec.so
+#path_noexec     : '/usr/libexec/sudo_noexec.so'
 
 ## include a directory containing multiple configuration files. These files
 ## can only contain default/user/group configuration. The global configuration will

lshell/checkconfig.py

@@ -93,7 +93,7 @@ def getoptions(self, arguments, conf):
             if f"{option[2:]}=" in variables.configparams:
                 conf[option[2:]] = value
             if option in ["-c"]:
-                conf["ssh"] = value
+                conf["ssh"] = value.strip()
             if option in ["-h", "--help"]:
                 utils.usage(exitcode=0)
             if option in ["--version"]:
@@ -694,26 +694,7 @@ def set_noexec(self):
                     break
 
         # in case the library was found, set the LD_PRELOAD aliases
-        if "path_noexec" in self.conf:
-            # exclude allowed_shell_escape commands from loop
-            exclude_se = list(
-                set(self.conf["allowed"])
-                - set(self.conf["allowed_shell_escape"])
-                - set(variables.builtins_list)
-            )
-
-            for cmd in exclude_se:
-                # take already set aliases into consideration
-                if cmd in self.conf["aliases"]:
-                    cmd = self.conf["aliases"][cmd]
-
-                # add an alias to all the commands, prepending with LD_PRELOAD=
-                # except for built-in commands
-                if cmd not in variables.builtins_list:
-                    self.conf["aliases"][
-                        cmd
-                    ] = f"LD_PRELOAD={self.conf['path_noexec']} {cmd}"
-        else:
+        if not self.conf.get("path_noexec"):
             # if sudo_noexec.so file is not found,  write error in log file,
             # but don't exit tp  prevent strict dependency on sudo noexec lib
             self.log.error("Error: noexec library not found")

lshell/shellcmd.py

@@ -70,6 +70,9 @@ def __init__(
         # initialize return code
         self.retcode = 0
 
+        # run overssh, if needed
+        self.run_overssh()
+
     def __getattr__(self, attr):
         """This method actually takes care of all the called method that are
         not resolved (i.e not existing methods). It actually will simulate
@@ -93,8 +96,6 @@ def __getattr__(self, attr):
             self.conf["promptprint"] = utils.updateprompt(os.getcwd(), self.conf)
             self.log = self.conf["logpath"]
 
-        self.check_scp_sftp()
-
         if self.g_cmd in ["quit", "exit", "EOF"]:
             self.do_exit()
 
@@ -193,7 +194,7 @@ def __getattr__(self, attr):
             self.mytimer(self.conf["timer"])
         return object.__getattribute__(self, attr)
 
-    def check_scp_sftp(self):
+    def run_overssh(self):
         """This method checks if the user is trying to SCP a file onto the
         server. If this is the case, it checks if the user is allowed to use
         SCP or not, and    acts as requested. : )
@@ -212,8 +213,6 @@ def check_scp_sftp(self):
                         self.log.error("*** forbidden SFTP connection")
                         sys.exit(1)
 
-                # initialize cli session
-                cli = ShellCmd(self.conf, None, None, None, None, self.conf["ssh"])
                 ret_check_path, self.conf = sec.check_path(
                     self.conf["ssh"], self.conf, ssh=1
                 )
@@ -254,7 +253,7 @@ def check_scp_sftp(self):
                                     f'SCP: upload forbidden: "{self.conf["ssh"]}"'
                                 )
                                 sys.exit(1)
-                        retcode = utils.cmd_parse_execute(self.conf["ssh"], cli)
+                        retcode = utils.cmd_parse_execute(self.conf["ssh"], self)
                         self.log.error("SCP disconnect")
                         sys.exit(retcode)
                     else:
@@ -272,14 +271,15 @@ def check_scp_sftp(self):
                     )
                     if ret_check_secure:
                         self.ssh_warn("char/command over SSH", self.conf["ssh"])
-                    # else
-                    self.log.error(f'Over SSH: "{self.conf["ssh"]}"')
+                        sys.exit(1)
+                    else:
+                        self.log.error(f'Over SSH: "{self.conf["ssh"]}"')
                     # if command is "help"
                     if self.conf["ssh"] == "help":
-                        cli.do_help(None)
+                        self.do_help(None)
                         retcode = 0
                     else:
-                        retcode = utils.cmd_parse_execute(self.conf["ssh"], cli)
+                        retcode = utils.cmd_parse_execute(self.conf["ssh"], self)
                     self.log.error("Exited")
                     sys.exit(retcode)
 
@@ -290,6 +290,7 @@ def check_scp_sftp(self):
             else:
                 # case of shell escapes
                 self.ssh_warn("shell escape", self.conf["ssh"])
+            return retcode
 
     def ssh_warn(self, message, command="", key=""):
         """log and warn if forbidden action over SSH"""

lshell/utils.py

@@ -103,6 +103,8 @@ def cmd_parse_execute(command_line, shell_context=None):
             else:
                 retcode = getattr(builtincmd, executable)(shell_context.conf)
         else:
+            if "path_noexec" in shell_context.conf:
+                command = f"LD_PRELOAD={shell_context.conf['path_noexec']} {command}"
             retcode = exec_cmd(command)
 
     return retcode

lshell/variables.py

@@ -3,7 +3,7 @@
 import sys
 import os
 
-__version__ = "0.10.4"
+__version__ = "0.10.5"
 
 # Required config variable list per user
 required_config = ["allowed", "forbidden", "warning_counter"]

man/lshell.1

@@ -1,7 +1,7 @@
 .\"
 .\"   Man page for the Limited Shell (lshell) project.
 .\"
-.TH lshell 1 "October, 2024" "v0.10.4"
+.TH lshell 1 "October, 2024" "v0.10.5"
 
 .SH NAME
 lshell \- Limited Shell

test/test_functional.py

@@ -815,3 +815,75 @@ def test_44_multicmd_asd_should_pass(self):
         self.child.expect(f"{self.user}:~\\$")
         result = self.child.before.decode("utf8").split("\n")[1].strip()
         self.assertEqual(expected, result)
+
+    def test_45_overssh_allowed_command_exit_0(self):
+        """F44 | Test 'ssh -c ls' command should exit 0"""
+        # add SSH_CLIENT to environment
+        if not os.environ.get("SSH_CLIENT"):
+            os.environ["SSH_CLIENT"] = "random"
+
+        self.child = pexpect.spawn(
+            f"{TOPDIR}/bin/lshell "
+            f"--config {TOPDIR}/etc/lshell.conf "
+            f"--overssh \"['ls']\" "
+            f"-c 'ls'"
+        )
+        self.child.expect(pexpect.EOF)
+
+        # Assert that the process exited
+        self.assertIsNotNone(
+            self.child.exitstatus, "The lshell process did not exit as expected."
+        )
+
+        # Optionally, you can assert that the exit code is correct
+        self.assertEqual(
+            self.child.exitstatus, 0, "The process should exit with code 1."
+        )
+
+    def test_46_overssh_allowed_command_exit_1(self):
+        """F44 | Test 'ssh -c ls' command should exit 1"""
+        # add SSH_CLIENT to environment
+        if not os.environ.get("SSH_CLIENT"):
+            os.environ["SSH_CLIENT"] = "random"
+
+        self.child = pexpect.spawn(
+            f"{TOPDIR}/bin/lshell "
+            f"--config {TOPDIR}/etc/lshell.conf "
+            f"--overssh \"['ls']\" "
+            f"-c 'ls /random'"
+        )
+        self.child.expect(pexpect.EOF)
+
+        # Assert that the process exited
+        self.assertIsNotNone(
+            self.child.exitstatus, "The lshell process did not exit as expected."
+        )
+
+        # Optionally, you can assert that the exit code is correct
+        self.assertEqual(
+            self.child.exitstatus, 1, "The process should exit with code 1."
+        )
+
+    def test_46_overssh_not_allowed_command_exit_1(self):
+        """F44 | Test 'ssh -c lss' command should succeed"""
+        # add SSH_CLIENT to environment
+        if not os.environ.get("SSH_CLIENT"):
+            os.environ["SSH_CLIENT"] = "random"
+
+        self.child = pexpect.spawn(
+            f"{TOPDIR}/bin/lshell "
+            f"--config {TOPDIR}/etc/lshell.conf "
+            f"--overssh \"['ls']\" "
+            f"-c 'lss'"
+        )
+        self.child.expect(pexpect.EOF)
+
+        # Assert that the process exited
+        self.assertIsNotNone(
+            self.child.exitstatus, "The lshell process did not exit as expected."
+        )
+
+        # Optionally, you can assert that the exit code is correct
+        self.assertEqual(
+            self.child.exitstatus, 1, "The process should exit with code 1."
+        )

test/test_unit.py

@@ -84,18 +84,6 @@ def test_11_checkconfig_configoverwrite(self):
         userconf = CheckConfig(args).returnconf()
         return self.assertEqual(userconf["strict"], 123)
 
-    def test_12_overssh(self):
-        """U12 | command over ssh"""
-        args = self.args + ["--overssh=['exit']", "-c exit"]
-        userconf = CheckConfig(args).returnconf()
-        shell = ShellCmd(userconf, args)
-        os.environ["SSH_CLIENT"] = "8.8.8.8 36000 22"
-        if "SSH_TTY" in os.environ:
-            os.environ.pop("SSH_TTY")
-        with self.assertRaises(SystemExit) as cm:
-            shell.check_scp_sftp()
-        return self.assertEqual(cm.exception.code, 0)
-
     def test_13_multiple_aliases_with_separator(self):
         """U13 | multiple aliases using &&, || and ; separators"""
         # enable &, | and ; characters
@@ -118,15 +106,6 @@ def test_14_sudo_all_commands_expansion(self):
         allowed.sort()
         return self.assertEqual(allowed, userconf["sudo_commands"])
 
-    def test_15_allowed_ld_preload_cmd(self):
-        """U15 | all allowed commands should be prepended with LD_PRELOAD"""
-        args = self.args + ["--allowed=['echo','export']"]
-        userconf = CheckConfig(args).returnconf()
-        # sort lists to compare
-        return self.assertEqual(
-            userconf["aliases"]["echo"], f"LD_PRELOAD={userconf['path_noexec']} echo"
-        )
-
     def test_16_allowed_ld_preload_builtin(self):
         """U16 | builtin commands should NOT be prepended with LD_PRELOAD"""
         args = self.args + ["--allowed=['echo','export']"]