MCP SSH Orchestrator

"""Tests for policy engine.""" import pytest from mcp_ssh.policy import Policy @pytest.fixture def basic_policy(): """Basic policy configuration for testing.""" return { "limits": { "max_seconds": 60, "max_output_bytes": 1048576, "deny_substrings": ["rm -rf", "shutdown"], }, "network": { "allow_ips": [], "allow_cidrs": ["10.0.0.0/8"], "block_ips": ["10.10.10.10"], "block_cidrs": [], }, "rules": [ { "action": "allow", "aliases": ["*"], "tags": [], "commands": ["uptime*", "df -h*"], }, { "action": "allow", "aliases": [], "tags": ["staging"], "commands": ["systemctl restart *"], }, { "action": "deny", "aliases": ["prod-*"], "tags": [], "commands": ["rm *"], }, ], "overrides": { "aliases": { "prod-db-1": { "max_seconds": 30, } }, "tags": { "production": { "max_output_bytes": 524288, } }, }, } def test_is_allowed_safe_command(basic_policy): """Test allowing safe command.""" pol = Policy(basic_policy) assert pol.is_allowed("test1", [], "uptime") is True assert pol.is_allowed("test1", [], "df -h") is True def test_is_denied_no_match(basic_policy): """Test denying command that doesn't match any rule.""" pol = Policy(basic_policy) # No rule matches, deny by default assert pol.is_allowed("test1", [], "unknown-command") is False def test_is_allowed_tag_based(basic_policy): """Test allowing command based on tag.""" pol = Policy(basic_policy) assert pol.is_allowed("stg-web-1", ["staging"], "systemctl restart nginx") is True assert ( pol.is_allowed("prod-web-1", ["production"], "systemctl restart nginx") is False ) def test_is_denied_by_substring(basic_policy): """Test denying command by substring.""" pol = Policy(basic_policy) assert pol.is_allowed("test1", [], "rm -rf /tmp") is False assert pol.is_allowed("test1", [], "shutdown -h now") is False def test_is_denied_by_explicit_rule(basic_policy): """Test denying command by explicit rule.""" pol = Policy(basic_policy) assert pol.is_allowed("prod-web-1", [], "rm file.txt") is False def test_limits_global(basic_policy): """Test global limits.""" pol = Policy(basic_policy) limits = pol.limits_for("test1", []) assert limits["max_seconds"] == 60 assert limits["max_output_bytes"] == 1048576 def test_limits_alias_override(basic_policy): """Test alias-specific limit override.""" pol = Policy(basic_policy) limits = pol.limits_for("prod-db-1", []) assert limits["max_seconds"] == 30 # Overridden def test_limits_tag_override(basic_policy): """Test tag-specific limit override.""" pol = Policy(basic_policy) limits = pol.limits_for("test1", ["production"]) assert limits["max_output_bytes"] == 524288 # Overridden def test_is_ip_allowed_in_cidr(basic_policy): """Test IP allowed by CIDR.""" pol = Policy(basic_policy) assert pol.is_ip_allowed("10.0.0.1") is True assert pol.is_ip_allowed("10.255.255.255") is True def test_is_ip_denied_outside_cidr(basic_policy): """Test IP denied outside allowed CIDR.""" pol = Policy(basic_policy) assert pol.is_ip_allowed("192.168.1.1") is False assert pol.is_ip_allowed("11.0.0.1") is False def test_is_ip_blocked_explicit(basic_policy): """Test IP explicitly blocked.""" pol = Policy(basic_policy) assert pol.is_ip_allowed("10.10.10.10") is False def test_is_ip_allowed_no_allowlist(): """Test IP allowed when no allowlist configured.""" pol = Policy({"network": {"allow_ips": [], "allow_cidrs": []}}) # No allowlist = allow all (unless blocked) assert pol.is_ip_allowed("1.2.3.4") is True assert pol.is_ip_allowed("10.0.0.1") is True def test_is_ip_allowed_exact_ip(): """Test exact IP match in allowlist.""" pol = Policy( { "network": { "allow_ips": ["10.0.0.1", "10.0.0.2"], "allow_cidrs": [], } } ) assert pol.is_ip_allowed("10.0.0.1") is True assert pol.is_ip_allowed("10.0.0.2") is True assert pol.is_ip_allowed("10.0.0.3") is False def test_require_known_host_default(basic_policy): """Test default known_host requirement.""" pol = Policy(basic_policy) # Should default to True assert pol.require_known_host() is True def test_require_known_host_network_override(): """Test known_host requirement from network config.""" pol = Policy( { "network": {"require_known_host": False}, "limits": {"require_known_host": True}, } ) # Network setting overrides limits assert pol.require_known_host() is False def test_glob_pattern_matching(): """Test glob pattern matching in rules.""" pol = Policy( { "rules": [ { "action": "allow", "aliases": ["web-*"], "tags": [], "commands": ["systemctl status *"], } ] } ) assert pol.is_allowed("web-1", [], "systemctl status nginx") is True assert pol.is_allowed("web-prod", [], "systemctl status apache") is True assert pol.is_allowed("db-1", [], "systemctl status nginx") is False def test_multiple_tags_any_match(): """Test that any tag matching enables the rule.""" pol = Policy( { "rules": [ { "action": "allow", "aliases": [], "tags": ["staging", "dev"], "commands": ["deploy*"], } ] } ) assert pol.is_allowed("test1", ["staging"], "deploy.sh") is True assert pol.is_allowed("test2", ["dev"], "deploy.sh") is True assert pol.is_allowed("test3", ["staging", "other"], "deploy.sh") is True assert pol.is_allowed("test4", ["production"], "deploy.sh") is False def test_empty_config(): """Test policy with empty config.""" pol = Policy({}) # Should have sensible defaults limits = pol.limits_for("test1", []) assert "max_seconds" in limits assert "max_output_bytes" in limits # Deny by default assert pol.is_allowed("test1", [], "any command") is False def test_bypass_prevention_quotes_single(basic_policy): """Test that single quotes don't bypass command denial.""" pol = Policy(basic_policy) # Should be blocked even with quotes assert pol.is_allowed("test1", [], "echo 'rm -rf /'") is False assert pol.is_allowed("test1", [], "'rm -rf /'") is False def test_bypass_prevention_quotes_double(basic_policy): """Test that double quotes don't bypass command denial.""" pol = Policy(basic_policy) # Should be blocked even with quotes assert pol.is_allowed("test1", [], 'echo "rm -rf /"') is False assert pol.is_allowed("test1", [], '"rm -rf /"') is False def test_bypass_prevention_mixed_quotes(basic_policy): """Test that mixed quotes don't bypass command denial.""" pol = Policy(basic_policy) # Should be blocked even with mixed quotes assert pol.is_allowed("test1", [], "echo \"rm -rf /'test'\"") is False assert pol.is_allowed("test1", [], "echo 'rm -rf /\"/test\"'") is False def test_bypass_prevention_escaped_characters(basic_policy): """Test that escaped characters don't bypass command denial.""" pol = Policy(basic_policy) # Should be blocked even with escaped characters assert pol.is_allowed("test1", [], "rm\\ -rf\\ /") is False assert pol.is_allowed("test1", [], "shutdown\\ -h") is False def test_bypass_prevention_whitespace_variations(basic_policy): """Test that whitespace variations don't bypass command denial.""" pol = Policy(basic_policy) # Multiple spaces should normalize to single space assert pol.is_allowed("test1", [], "rm -rf /") is False # Tabs should normalize to space assert pol.is_allowed("test1", [], "rm\t-rf\t/") is False # Mixed whitespace should normalize assert pol.is_allowed("test1", [], "rm \t -rf \t /") is False def test_bypass_prevention_token_matching(basic_policy): """Test that token-based matching prevents bypasses.""" # Token-based matching is tested via normalized command checking # which checks both original and normalized forms # Additional token-specific tests would require more complex policy rules pass def test_bypass_prevention_legitimate_commands_still_work(basic_policy): """Test that legitimate commands still work despite normalization.""" pol = Policy(basic_policy) # These should still be allowed (match allow rules, don't match deny_substrings) assert pol.is_allowed("test1", [], "uptime") is True assert pol.is_allowed("test1", [], "df -h") is True # Commands with quotes that match allow rules should still work # Note: 'echo "hello world"' doesn't match any allow rule, so would be denied # But commands that ARE allowed by rules should work even with quotes assert ( pol.is_allowed("test1", [], 'uptime "server1"') is True ) # Still contains "uptime" def test_normalize_command_quotes_removal(): """Test command normalization removes quotes.""" from mcp_ssh.policy import _normalize_command assert _normalize_command("rm -rf /") == "rm -rf /" assert _normalize_command('"rm -rf /"') == "rm -rf /" assert _normalize_command("'rm -rf /'") == "rm -rf /" assert _normalize_command('echo "rm -rf /"') == "echo rm -rf /" def test_normalize_command_escape_removal(): """Test command normalization removes escaping.""" from mcp_ssh.policy import _normalize_command assert _normalize_command("rm\\ -rf\\ /") == "rm -rf /" assert _normalize_command("shutdown\\ -h") == "shutdown -h" def test_normalize_command_whitespace_normalization(): """Test command normalization normalizes whitespace.""" from mcp_ssh.policy import _normalize_command assert _normalize_command("rm -rf /") == "rm -rf /" assert _normalize_command("rm\t-rf\t/") == "rm -rf /" assert _normalize_command("rm \t -rf \t /") == "rm -rf /" assert _normalize_command(" rm -rf / ") == "rm -rf /" def test_normalize_command_complex_bypass(): """Test command normalization handles complex bypass attempts.""" from mcp_ssh.policy import _normalize_command # Complex bypass attempt normalized = _normalize_command('echo "rm\\ -rf\\ /"') # Should normalize to: echo rm -rf / assert "rm -rf /" in normalized assert '"' not in normalized assert "\\" not in normalized # Another complex case normalized = _normalize_command("'rm' '-rf' '/'") assert normalized == "rm -rf /" def test_bypass_prevention_substring_in_normalized_only(basic_policy): """Test that commands blocked only in normalized form are detected.""" pol = Policy(basic_policy) # Command that would bypass without normalization # 'rm -rf /' doesn't contain 'rm -rf /' directly, but normalized does assert pol.is_allowed("test1", [], "'rm -rf /'") is False # Similar with escaped assert pol.is_allowed("test1", [], "rm\\ -rf\\ /") is False def test_bypass_prevention_multiple_patterns(basic_policy): """Test bypass prevention works for multiple deny patterns.""" pol = Policy(basic_policy) # Both patterns should be caught assert pol.is_allowed("test1", [], "'rm -rf /'") is False assert pol.is_allowed("test1", [], '"shutdown -h"') is False