diff --git a/amc_server/mixins/spawn.py b/amc_server/mixins/spawn.py index 7e67040..5021297 100644 --- a/amc_server/mixins/spawn.py +++ b/amc_server/mixins/spawn.py @@ -1,5 +1,6 @@ import json import os +import re import subprocess import time import uuid @@ -20,6 +21,20 @@ AGENT_COMMANDS = { # Module-level cache for projects list (AC-33) _projects_cache: list[str] = [] +# Characters unsafe for Zellij pane/tab names: control chars, quotes, backticks +_UNSAFE_PANE_CHARS = re.compile(r'[\x00-\x1f\x7f"\'`]') + + +def _sanitize_pane_name(name): + """Sanitize a string for use as a Zellij pane name. + + Replaces control characters and quotes with underscores, collapses runs + of whitespace into a single space, and truncates to 64 chars. + """ + name = _UNSAFE_PANE_CHARS.sub('_', name) + name = re.sub(r'\s+', ' ', name).strip() + return name[:64] if name else 'unnamed' + def load_projects_cache(): """Scan ~/projects/ and cache the list. Called on server start.""" @@ -112,9 +127,17 @@ class SpawnMixin: def _validate_spawn_params(self, project, agent_type): """Validate spawn parameters. Returns resolved_path or error dict.""" - if not project: + if not project or not isinstance(project, str): return {'error': 'Project name is required', 'code': 'MISSING_PROJECT'} + # Reject whitespace-only names + if not project.strip(): + return {'error': 'Project name is required', 'code': 'MISSING_PROJECT'} + + # Reject null bytes and control characters (U+0000-U+001F, U+007F) + if '\x00' in project or re.search(r'[\x00-\x1f\x7f]', project): + return {'error': 'Invalid project name', 'code': 'INVALID_PROJECT'} + # Reject path traversal characters (/, \, ..) if '/' in project or '\\' in project or '..' in project: return {'error': 'Invalid project name', 'code': 'INVALID_PROJECT'} @@ -221,7 +244,7 @@ class SpawnMixin: # Build agent command agent_cmd = AGENT_COMMANDS[agent_type] - pane_name = f'{agent_type}-{project}' + pane_name = _sanitize_pane_name(f'{agent_type}-{project}') # Spawn pane with agent command env = os.environ.copy() diff --git a/tests/test_spawn.py b/tests/test_spawn.py index 9d63696..5e5f752 100644 --- a/tests/test_spawn.py +++ b/tests/test_spawn.py @@ -7,7 +7,7 @@ from pathlib import Path from unittest.mock import MagicMock, patch import amc_server.mixins.spawn as spawn_mod -from amc_server.mixins.spawn import SpawnMixin, load_projects_cache, _projects_cache +from amc_server.mixins.spawn import SpawnMixin, load_projects_cache, _projects_cache, _sanitize_pane_name class DummySpawnHandler(SpawnMixin): @@ -602,5 +602,290 @@ class TestHandleHealth(unittest.TestCase): self.assertFalse(payload['zellij_available']) +# --------------------------------------------------------------------------- +# Special characters in project names (bd-14p) +# --------------------------------------------------------------------------- + +class TestSpecialCharacterValidation(unittest.TestCase): + """Verify project names with special characters are handled correctly.""" + + def setUp(self): + self.handler = DummySpawnHandler() + + def _make_project(self, tmpdir, name): + """Create a project directory with the given name and patch PROJECTS_DIR.""" + projects = Path(tmpdir) / 'projects' + projects.mkdir(exist_ok=True) + project_dir = projects / name + project_dir.mkdir() + return projects, project_dir + + # --- safe characters that should work --- + + def test_hyphenated_name(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'my-project') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('my-project', 'claude') + self.assertNotIn('error', result) + + def test_underscored_name(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'my_project') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('my_project', 'claude') + self.assertNotIn('error', result) + + def test_numeric_name(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'project2') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('project2', 'claude') + self.assertNotIn('error', result) + + def test_name_with_spaces(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'my project') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('my project', 'claude') + self.assertNotIn('error', result) + + def test_name_with_dots(self): + """Single dots are fine, only '..' is rejected.""" + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'my.project') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('my.project', 'claude') + self.assertNotIn('error', result) + + def test_name_with_parentheses(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'project (copy)') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('project (copy)', 'claude') + self.assertNotIn('error', result) + + def test_name_with_at_sign(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, '@scope-pkg') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('@scope-pkg', 'claude') + self.assertNotIn('error', result) + + def test_name_with_plus(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'c++project') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('c++project', 'claude') + self.assertNotIn('error', result) + + def test_name_with_hash(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'c#project') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('c#project', 'claude') + self.assertNotIn('error', result) + + # --- control characters: should be rejected --- + + def test_null_byte_rejected(self): + result = self.handler._validate_spawn_params('project\x00evil', 'claude') + self.assertEqual(result['code'], 'INVALID_PROJECT') + + def test_newline_rejected(self): + result = self.handler._validate_spawn_params('project\nevil', 'claude') + self.assertEqual(result['code'], 'INVALID_PROJECT') + + def test_tab_rejected(self): + result = self.handler._validate_spawn_params('project\tevil', 'claude') + self.assertEqual(result['code'], 'INVALID_PROJECT') + + def test_carriage_return_rejected(self): + result = self.handler._validate_spawn_params('project\revil', 'claude') + self.assertEqual(result['code'], 'INVALID_PROJECT') + + def test_escape_char_rejected(self): + result = self.handler._validate_spawn_params('project\x1bevil', 'claude') + self.assertEqual(result['code'], 'INVALID_PROJECT') + + def test_bell_char_rejected(self): + result = self.handler._validate_spawn_params('project\x07evil', 'claude') + self.assertEqual(result['code'], 'INVALID_PROJECT') + + def test_del_char_rejected(self): + result = self.handler._validate_spawn_params('project\x7fevil', 'claude') + self.assertEqual(result['code'], 'INVALID_PROJECT') + + # --- whitespace edge cases --- + + def test_whitespace_only_space(self): + result = self.handler._validate_spawn_params(' ', 'claude') + self.assertEqual(result['code'], 'MISSING_PROJECT') + + def test_whitespace_only_multiple_spaces(self): + result = self.handler._validate_spawn_params(' ', 'claude') + self.assertEqual(result['code'], 'MISSING_PROJECT') + + # --- non-string project name --- + + def test_integer_project_name_rejected(self): + result = self.handler._validate_spawn_params(42, 'claude') + self.assertEqual(result['code'], 'MISSING_PROJECT') + + def test_list_project_name_rejected(self): + result = self.handler._validate_spawn_params(['bad'], 'claude') + self.assertEqual(result['code'], 'MISSING_PROJECT') + + # --- shell metacharacters (safe because subprocess uses list args) --- + + def test_dollar_sign_in_name(self): + """Dollar sign is safe - no shell expansion with list args.""" + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, '$HOME') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('$HOME', 'claude') + self.assertNotIn('error', result) + + def test_semicolon_in_name(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'foo;bar') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('foo;bar', 'claude') + self.assertNotIn('error', result) + + def test_backtick_in_name(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'foo`bar') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('foo`bar', 'claude') + self.assertNotIn('error', result) + + def test_pipe_in_name(self): + with tempfile.TemporaryDirectory() as tmpdir: + projects, _ = self._make_project(tmpdir, 'foo|bar') + with patch.object(spawn_mod, 'PROJECTS_DIR', projects): + result = self.handler._validate_spawn_params('foo|bar', 'claude') + self.assertNotIn('error', result) + + +# --------------------------------------------------------------------------- +# _sanitize_pane_name (bd-14p) +# --------------------------------------------------------------------------- + +class TestSanitizePaneName(unittest.TestCase): + """Tests for Zellij pane name sanitization.""" + + def test_simple_name_unchanged(self): + self.assertEqual(_sanitize_pane_name('claude-myproject'), 'claude-myproject') + + def test_name_with_spaces_preserved(self): + self.assertEqual(_sanitize_pane_name('claude-my project'), 'claude-my project') + + def test_multiple_spaces_collapsed(self): + self.assertEqual(_sanitize_pane_name('claude-my project'), 'claude-my project') + + def test_quotes_replaced(self): + self.assertEqual(_sanitize_pane_name('claude-"quoted"'), 'claude-_quoted_') + + def test_single_quotes_replaced(self): + self.assertEqual(_sanitize_pane_name("claude-it's"), 'claude-it_s') + + def test_backtick_replaced(self): + self.assertEqual(_sanitize_pane_name('claude-`cmd`'), 'claude-_cmd_') + + def test_control_chars_replaced(self): + self.assertEqual(_sanitize_pane_name('claude-\x07bell'), 'claude-_bell') + + def test_tab_replaced(self): + self.assertEqual(_sanitize_pane_name('claude-\ttab'), 'claude-_tab') + + def test_truncated_at_64_chars(self): + long_name = 'a' * 100 + result = _sanitize_pane_name(long_name) + self.assertEqual(len(result), 64) + + def test_empty_returns_unnamed(self): + self.assertEqual(_sanitize_pane_name(''), 'unnamed') + + def test_only_control_chars_returns_underscores(self): + result = _sanitize_pane_name('\x01\x02\x03') + self.assertEqual(result, '___') + + def test_unicode_preserved(self): + self.assertEqual(_sanitize_pane_name('claude-cafe'), 'claude-cafe') + + def test_hyphens_and_underscores_preserved(self): + self.assertEqual(_sanitize_pane_name('claude-my_project-v2'), 'claude-my_project-v2') + + +# --------------------------------------------------------------------------- +# Special chars in spawn command construction (bd-14p) +# --------------------------------------------------------------------------- + +class TestSpawnWithSpecialChars(unittest.TestCase): + """Verify special character project names pass through to Zellij correctly.""" + + def test_space_in_project_name_passes_to_zellij(self): + """Project with spaces should be passed correctly to subprocess args.""" + handler = DummySpawnHandler() + + with patch.object(handler, '_check_zellij_session_exists', return_value=True), \ + patch.object(handler, '_wait_for_session_file', return_value=True), \ + patch('amc_server.mixins.spawn.subprocess.run') as mock_run: + mock_run.return_value = MagicMock(returncode=0, stderr='') + result = handler._spawn_agent_in_project_tab( + 'my project', Path('/fake/my project'), 'claude', 'spawn-123', + ) + + self.assertTrue(result['ok']) + # Check the tab creation call contains the name with spaces + tab_call = mock_run.call_args_list[0] + tab_args = tab_call[0][0] + self.assertIn('my project', tab_args) + + # Check pane name was sanitized (spaces preserved but other chars cleaned) + pane_call = mock_run.call_args_list[1] + pane_args = pane_call[0][0] + name_idx = pane_args.index('--name') + 1 + self.assertEqual(pane_args[name_idx], 'claude-my project') + + def test_quotes_in_project_name_sanitized_in_pane_name(self): + """Quotes in project names should be stripped from pane names.""" + handler = DummySpawnHandler() + + with patch.object(handler, '_check_zellij_session_exists', return_value=True), \ + patch.object(handler, '_wait_for_session_file', return_value=True), \ + patch('amc_server.mixins.spawn.subprocess.run') as mock_run: + mock_run.return_value = MagicMock(returncode=0, stderr='') + handler._spawn_agent_in_project_tab( + 'it\'s-a-project', Path('/fake/its-a-project'), 'claude', 'spawn-123', + ) + + pane_call = mock_run.call_args_list[1] + pane_args = pane_call[0][0] + name_idx = pane_args.index('--name') + 1 + # Single quote should be replaced with underscore + self.assertNotIn("'", pane_args[name_idx]) + self.assertEqual(pane_args[name_idx], 'claude-it_s-a-project') + + def test_unicode_project_name_in_pane(self): + """Unicode project names should pass through to pane names.""" + handler = DummySpawnHandler() + + with patch.object(handler, '_check_zellij_session_exists', return_value=True), \ + patch.object(handler, '_wait_for_session_file', return_value=True), \ + patch('amc_server.mixins.spawn.subprocess.run') as mock_run: + mock_run.return_value = MagicMock(returncode=0, stderr='') + result = handler._spawn_agent_in_project_tab( + 'cafe', Path('/fake/cafe'), 'claude', 'spawn-123', + ) + + self.assertTrue(result['ok']) + pane_call = mock_run.call_args_list[1] + pane_args = pane_call[0][0] + name_idx = pane_args.index('--name') + 1 + self.assertEqual(pane_args[name_idx], 'claude-cafe') + + if __name__ == '__main__': unittest.main()