From 68d26fc01dd04fd928b492f9fd9952b603f1c222 Mon Sep 17 00:00:00 2001 From: sshrushanth-ks Date: Wed, 23 Sep 2026 13:42:32 +0530 Subject: [PATCH 1/2] Fix: Prevent newline injection in connect command prompts Escape control chars and handle comments to prevent hidden commands from bypassing the confirmation prompt. Attackers with Can Edit on shared records could exfiltrate device credentials without operator visibility. --- keepercommander/commands/connect_prompts.py | 39 +++++++++++++-- unit-tests/test_connect_security.py | 55 +++++++++++++++++++-- 2 files changed, 86 insertions(+), 8 deletions(-) diff --git a/keepercommander/commands/connect_prompts.py b/keepercommander/commands/connect_prompts.py index d3cd66760..5d48a21a6 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,14 @@ 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 (#) — everything after # until newline is ignored + """ if not script: return [] @@ -174,6 +202,11 @@ def flush() -> None: elif c in ('"', "'"): quote = c buf.append(c) + elif c == '#' and quote is None: + # Comment: skip until newline + 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..21f9a2367 100644 --- a/unit-tests/test_connect_security.py +++ b/unit-tests/test_connect_security.py @@ -100,14 +100,59 @@ 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 "#"') + 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__': From 6b8b08d735f5cd6db2f7d4baecfce4f7ca1f6c1c Mon Sep 17 00:00:00 2001 From: sshrushanth-ks Date: Wed, 23 Sep 2026 14:08:26 +0530 Subject: [PATCH 2/2] Add word-boundary check for shell comments; cover with 3 regression tests --- keepercommander/commands/connect_prompts.py | 14 ++++++++++--- unit-tests/test_connect_security.py | 23 +++++++++++++++++++++ 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/keepercommander/commands/connect_prompts.py b/keepercommander/commands/connect_prompts.py index 5d48a21a6..0b023f10e 100644 --- a/keepercommander/commands/connect_prompts.py +++ b/keepercommander/commands/connect_prompts.py @@ -170,7 +170,8 @@ def split_shell_statements(script: Optional[str]) -> List[str]: - Single/double quoted strings (quote escaping) - Escaped characters (backslash) - Shell statement separators (; \n && ||) - - Comments (#) — everything after # until newline is ignored + - Comments (#) — # only starts a comment when it begins a word + (at line start or preceded by whitespace) """ if not script: return [] @@ -187,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: @@ -202,8 +210,8 @@ def flush() -> None: elif c in ('"', "'"): quote = c buf.append(c) - elif c == '#' and quote is None: - # Comment: skip until newline + 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 diff --git a/unit-tests/test_connect_security.py b/unit-tests/test_connect_security.py index 21f9a2367..b8c50842c 100644 --- a/unit-tests/test_connect_security.py +++ b/unit-tests/test_connect_security.py @@ -122,6 +122,29 @@ def test_quoted_hash_is_not_comment(self): 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_escaped(self):