"""
Safety mode implementation for kubectl-mcp-server.
Provides read-only and disable-destructive modes to prevent accidental cluster mutations.
"""
from enum import Enum
from typing import Any, Dict, Set
import logging
logger = logging.getLogger("mcp-server")
class SafetyMode(Enum):
"""Safety mode levels for the MCP server."""
NORMAL = "normal"
READ_ONLY = "read_only"
DISABLE_DESTRUCTIVE = "disable_destructive"
# Global safety mode state
_current_mode: SafetyMode = SafetyMode.NORMAL
# Operations that modify cluster state (blocked in READ_ONLY mode)
WRITE_OPERATIONS: Set[str] = {
# Pod operations
"run_pod", "delete_pod",
# Deployment operations
"scale_deployment", "restart_deployment", "delete_deployment",
"rollback_deployment", "create_deployment", "update_deployment",
# StatefulSet operations
"scale_statefulset", "restart_statefulset", "delete_statefulset",
# DaemonSet operations
"restart_daemonset", "delete_daemonset",
# Service operations
"create_service", "delete_service", "update_service",
# ConfigMap/Secret operations
"create_configmap", "delete_configmap", "update_configmap",
"create_secret", "delete_secret", "update_secret",
# Namespace operations
"create_namespace", "delete_namespace",
# Helm operations
"install_helm_chart", "upgrade_helm_chart", "uninstall_helm_chart",
"rollback_helm_release",
# kubectl operations
"apply_manifest", "delete_resource", "patch_resource",
"create_resource", "replace_resource",
# Context operations
"switch_context", "set_namespace_for_context",
# Rollout operations
"rollout_promote_tool", "rollout_abort_tool", "rollout_retry_tool",
"rollout_restart_tool",
# KubeVirt operations
"kubevirt_vm_start_tool", "kubevirt_vm_stop_tool", "kubevirt_vm_restart_tool",
"kubevirt_vm_pause_tool", "kubevirt_vm_unpause_tool", "kubevirt_vm_migrate_tool",
# CAPI operations
"capi_machinedeployment_scale_tool",
}
# Operations that are destructive (blocked in DISABLE_DESTRUCTIVE mode)
DESTRUCTIVE_OPERATIONS: Set[str] = {
# Delete operations
"delete_pod", "delete_deployment", "delete_statefulset", "delete_daemonset",
"delete_service", "delete_configmap", "delete_secret", "delete_namespace",
"delete_resource",
# Helm uninstall
"uninstall_helm_chart",
# Rollout abort
"rollout_abort_tool",
# VM stop
"kubevirt_vm_stop_tool",
}
def get_safety_mode() -> SafetyMode:
"""Get the current safety mode."""
return _current_mode
def set_safety_mode(mode: SafetyMode) -> None:
"""Set the safety mode globally."""
global _current_mode
_current_mode = mode
logger.info(f"Safety mode set to: {mode.value}")
def get_mode_info() -> Dict[str, Any]:
"""Get information about the current safety mode."""
mode = get_safety_mode()
return {
"mode": mode.value,
"description": {
SafetyMode.NORMAL: "All operations allowed",
SafetyMode.READ_ONLY: "Only read operations allowed (no create/update/delete)",
SafetyMode.DISABLE_DESTRUCTIVE: "Create/update allowed, delete operations blocked",
}[mode],
"blocked_operations": {
SafetyMode.NORMAL: [],
SafetyMode.READ_ONLY: sorted(WRITE_OPERATIONS | DESTRUCTIVE_OPERATIONS),
SafetyMode.DISABLE_DESTRUCTIVE: sorted(DESTRUCTIVE_OPERATIONS),
}[mode]
}
