managing contexts

README.md

@@ -5,7 +5,7 @@
 ## Особенности
 
 - 🎯 Интерактивное меню с навигацией стрелками
-- 🔄 Переключение между Kubernetes контекстами/кластерами
+- 🔄 Полное управление Kubernetes контекстами (просмотр, переключение, добавление, удаление)
 - ⭐ Избранные namespaces для быстрого доступа
 - 🎨 Красивый вывод с использованием Rich
 - 📦 Управление namespaces, deployments, pods, ConfigMaps
@@ -13,6 +13,7 @@
 - 📝 Просмотр логов pods
 - 🔍 Просмотр и редактирование ConfigMaps в текстовом редакторе
 - 💾 Сохранение настроек в `~/.config/k8s-tool/k8s-tool.cfg`
+- 🔒 Автоматическое создание резервных копий kubeconfig при изменениях
 
 ## Требования
 
@@ -58,14 +59,29 @@ poetry run k8s-tool
 
 ### Основные функции
 
-#### Переключение контекста Kubernetes
-Быстрое переключение между различными Kubernetes кластерами/контекстами:
+#### Управление контекстами Kubernetes
+Полное управление Kubernetes контекстами/кластерами:
+
+**Просмотр и переключение:**
 - Просмотр всех доступных контекстов в виде таблицы
 - Отображение текущего активного контекста (✓)
 - Переключение на другой контекст
 - Автоматическая переинициализация подключения к новому кластеру
 - После переключения контекста namespace сбрасывается и нужно выбрать заново
 
+**Добавление нового контекста:**
+- Интерактивное создание нового контекста
+- Указание имени контекста, кластера, пользователя
+- API server URL
+- Опционально: токен аутентификации
+- Опционально: сертификат CA (base64)
+- Автоматическое создание резервной копии kubeconfig перед изменением
+
+**Удаление контекста:**
+- Удаление ненужных контекстов
+- Защита от удаления текущего активного контекста
+- Автоматическое создание резервной копии kubeconfig
+
 #### Выбор Namespace
 Выберите namespace для работы. Все последующие операции будут выполняться в выбранном namespace.
 

k8s_tool/k8s_client.py

@@ -1,6 +1,9 @@
 """Kubernetes API client wrapper."""
 
+import os
+import yaml
 from typing import List, Dict, Any, Optional
+from pathlib import Path
 from kubernetes import client, config
 from kubernetes.client.exceptions import ApiException
 from rich.console import Console
@@ -65,6 +68,144 @@ class K8sClient:
             console.print(f"[red]Error switching context:[/red] {e}")
             return False
 
+    def _get_kubeconfig_path(self) -> Path:
+        """Get path to kubeconfig file."""
+        kubeconfig = os.environ.get('KUBECONFIG')
+        if kubeconfig:
+            return Path(kubeconfig)
+        return Path.home() / '.kube' / 'config'
+
+    def _load_kubeconfig(self) -> Optional[Dict]:
+        """Load kubeconfig file."""
+        try:
+            kubeconfig_path = self._get_kubeconfig_path()
+            if not kubeconfig_path.exists():
+                console.print(f"[red]Kubeconfig not found:[/red] {kubeconfig_path}")
+                return None
+
+            with open(kubeconfig_path, 'r') as f:
+                return yaml.safe_load(f)
+        except Exception as e:
+            console.print(f"[red]Error loading kubeconfig:[/red] {e}")
+            return None
+
+    def _save_kubeconfig(self, kubeconfig: Dict) -> bool:
+        """Save kubeconfig file."""
+        try:
+            kubeconfig_path = self._get_kubeconfig_path()
+            # Create backup
+            backup_path = kubeconfig_path.with_suffix('.backup')
+            if kubeconfig_path.exists():
+                import shutil
+                shutil.copy(kubeconfig_path, backup_path)
+
+            with open(kubeconfig_path, 'w') as f:
+                yaml.dump(kubeconfig, f, default_flow_style=False)
+
+            return True
+        except Exception as e:
+            console.print(f"[red]Error saving kubeconfig:[/red] {e}")
+            return False
+
+    def delete_context(self, context_name: str) -> bool:
+        """Delete a context from kubeconfig."""
+        try:
+            kubeconfig = self._load_kubeconfig()
+            if not kubeconfig:
+                return False
+
+            # Check if context exists
+            contexts = kubeconfig.get('contexts', [])
+            context_names = [ctx['name'] for ctx in contexts]
+
+            if context_name not in context_names:
+                console.print(f"[yellow]Context '{context_name}' not found[/yellow]")
+                return False
+
+            # Don't allow deleting current context
+            if context_name == self.current_context:
+                console.print(f"[yellow]Cannot delete current context. Switch to another context first.[/yellow]")
+                return False
+
+            # Remove context
+            kubeconfig['contexts'] = [ctx for ctx in contexts if ctx['name'] != context_name]
+
+            # Save updated config
+            if self._save_kubeconfig(kubeconfig):
+                console.print(f"[green]✓[/green] Context '{context_name}' deleted")
+                return True
+
+            return False
+        except Exception as e:
+            console.print(f"[red]Error deleting context:[/red] {e}")
+            return False
+
+    def add_context(self, context_name: str, cluster_name: str, user_name: str,
+                   server: str, certificate_authority: Optional[str] = None,
+                   token: Optional[str] = None) -> bool:
+        """Add a new context to kubeconfig."""
+        try:
+            kubeconfig = self._load_kubeconfig()
+            if not kubeconfig:
+                return False
+
+            # Check if context already exists
+            contexts = kubeconfig.get('contexts', [])
+            if any(ctx['name'] == context_name for ctx in contexts):
+                console.print(f"[yellow]Context '{context_name}' already exists[/yellow]")
+                return False
+
+            # Add cluster if it doesn't exist
+            clusters = kubeconfig.get('clusters', [])
+            if not any(cls['name'] == cluster_name for cls in clusters):
+                new_cluster = {
+                    'name': cluster_name,
+                    'cluster': {
+                        'server': server
+                    }
+                }
+                if certificate_authority:
+                    new_cluster['cluster']['certificate-authority-data'] = certificate_authority
+                else:
+                    new_cluster['cluster']['insecure-skip-tls-verify'] = True
+
+                clusters.append(new_cluster)
+                kubeconfig['clusters'] = clusters
+
+            # Add user if it doesn't exist
+            users = kubeconfig.get('users', [])
+            if not any(usr['name'] == user_name for usr in users):
+                new_user = {
+                    'name': user_name,
+                    'user': {}
+                }
+                if token:
+                    new_user['user']['token'] = token
+
+                users.append(new_user)
+                kubeconfig['users'] = users
+
+            # Add context
+            new_context = {
+                'name': context_name,
+                'context': {
+                    'cluster': cluster_name,
+                    'user': user_name
+                }
+            }
+            contexts.append(new_context)
+            kubeconfig['contexts'] = contexts
+
+            # Save updated config
+            if self._save_kubeconfig(kubeconfig):
+                console.print(f"[green]✓[/green] Context '{context_name}' added")
+                return True
+
+            return False
+        except Exception as e:
+            console.print(f"[red]Error adding context:[/red] {e}")
+            return False
+
     def get_namespaces(self) -> List[str]:
         """Get list of all namespaces."""
         try:

k8s_tool/main.py

@@ -74,7 +74,7 @@ class K8sTool:
         console.print(f"[bold]Current namespace:[/bold] {namespace_info}{fav_indicator}")
 
         choices = [
-            "Switch Context",
+            "Manage Contexts",
             "Select Namespace",
             "Manage Favorites",
             "List Deployments",
@@ -92,8 +92,8 @@ class K8sTool:
             style=custom_style
         ).ask()
 
-        if action == "Switch Context":
-            self._switch_context()
+        if action == "Manage Contexts":
+            self._manage_contexts()
         elif action == "Select Namespace":
             self._select_namespace()
         elif action == "Manage Favorites":
@@ -152,6 +152,45 @@ class K8sTool:
             fav_text = " (favorite)" if is_fav else ""
             console.print(f"[green]✓[/green] Namespace set to: [cyan]{namespace}[/cyan]{fav_text}")
 
+    def _manage_contexts(self):
+        """Manage Kubernetes contexts."""
+        choices = [
+            "View all contexts",
+            "Switch context",
+            "Add new context",
+            "Delete context",
+            "Back to main menu"
+        ]
+
+        action = questionary.select(
+            "Context management:",
+            choices=choices,
+            style=custom_style
+        ).ask()
+
+        if not action or action == "Back to main menu":
+            return
+
+        if action == "View all contexts":
+            self._view_contexts()
+        elif action == "Switch context":
+            self._switch_context()
+        elif action == "Add new context":
+            self._add_context()
+        elif action == "Delete context":
+            self._delete_context()
+
+    def _view_contexts(self):
+        """View all available contexts."""
+        console.print("[dim]Fetching contexts...[/dim]")
+        contexts = self.k8s_client.get_contexts()
+
+        if not contexts:
+            console.print("[red]No contexts found[/red]")
+            return
+
+        self.k8s_client.display_contexts_table(contexts)
+
     def _switch_context(self):
         """Switch Kubernetes context."""
         console.print("[dim]Fetching contexts...[/dim]")
@@ -171,7 +210,7 @@ class K8sTool:
             console.print("[yellow]Only one context available[/yellow]")
             return
 
-        # Add option to view current
+        # Add option to cancel
         choices = available_contexts + [questionary.Separator(), "Cancel"]
 
         selected = questionary.select(
@@ -186,6 +225,117 @@ class K8sTool:
                 self.current_namespace = None
                 console.print("[yellow]Note:[/yellow] Namespace selection reset. Please select a namespace.")
 
+    def _add_context(self):
+        """Add a new Kubernetes context."""
+        console.print("\n[bold]Add New Context[/bold]\n")
+
+        # Get context details
+        context_name = questionary.text(
+            "Context name:",
+            style=custom_style
+        ).ask()
+
+        if not context_name:
+            return
+
+        cluster_name = questionary.text(
+            "Cluster name:",
+            default=context_name,
+            style=custom_style
+        ).ask()
+
+        if not cluster_name:
+            return
+
+        server = questionary.text(
+            "API server URL (e.g., https://kubernetes.example.com:6443):",
+            style=custom_style
+        ).ask()
+
+        if not server:
+            return
+
+        user_name = questionary.text(
+            "User name:",
+            default=context_name + "-user",
+            style=custom_style
+        ).ask()
+
+        if not user_name:
+            return
+
+        # Optional: token
+        has_token = questionary.confirm(
+            "Do you have an authentication token?",
+            style=custom_style,
+            default=False
+        ).ask()
+
+        token = None
+        if has_token:
+            token = questionary.password(
+                "Authentication token:",
+                style=custom_style
+            ).ask()
+
+        # Optional: certificate
+        has_cert = questionary.confirm(
+            "Do you have a certificate authority data?",
+            style=custom_style,
+            default=False
+        ).ask()
+
+        cert = None
+        if has_cert:
+            cert = questionary.text(
+                "Certificate authority data (base64 encoded):",
+                style=custom_style
+            ).ask()
+
+        # Confirm
+        console.print("\n[bold]Review:[/bold]")
+        console.print(f"  Context: [cyan]{context_name}[/cyan]")
+        console.print(f"  Cluster: [cyan]{cluster_name}[/cyan]")
+        console.print(f"  Server: [cyan]{server}[/cyan]")
+        console.print(f"  User: [cyan]{user_name}[/cyan]")
+        console.print(f"  Token: [cyan]{'***' if token else 'None'}[/cyan]")
+        console.print(f"  Certificate: [cyan]{'***' if cert else 'None (insecure)'}[/cyan]")
+
+        if questionary.confirm("\nAdd this context?", style=custom_style, default=False).ask():
+            self.k8s_client.add_context(context_name, cluster_name, user_name, server, cert, token)
+
+    def _delete_context(self):
+        """Delete a Kubernetes context."""
+        console.print("[dim]Fetching contexts...[/dim]")
+        contexts = self.k8s_client.get_contexts()
+
+        if not contexts:
+            console.print("[red]No contexts found[/red]")
+            return
+
+        # Filter out current context
+        deletable_contexts = [ctx['name'] for ctx in contexts if not ctx['is_active']]
+
+        if not deletable_contexts:
+            console.print("[yellow]No contexts available for deletion (cannot delete current context)[/yellow]")
+            return
+
+        context_name = questionary.select(
+            "Select context to delete:",
+            choices=deletable_contexts + [questionary.Separator(), "Cancel"],
+            style=custom_style
+        ).ask()
+
+        if not context_name or context_name == "Cancel":
+            return
+
+        if questionary.confirm(
+            f"Are you sure you want to delete context '{context_name}'?",
+            style=custom_style,
+            default=False
+        ).ask():
+            self.k8s_client.delete_context(context_name)
+
     def _manage_favorites(self):
         """Manage favorite namespaces."""
         choices = []