diff --git a/.gitignore b/.gitignore index 5a5cbfa..1d32645 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,7 @@ .DS_Store .vscode .idea +CLAUDE.md # App artifacts .kubebin diff --git a/Pipfile b/Pipfile index 39c7111..ad37312 100644 --- a/Pipfile +++ b/Pipfile @@ -16,4 +16,4 @@ kubectl-application-shell = {file = ".", editable = true} python_version = "3.9" [scripts] -kubeas = {call = "kubectl_application_shell.cli:app()"} +kubeas = {call = "kubectl_application_shell:run_cli()"} diff --git a/README.md b/README.md index 7bda589..ffe9edc 100644 --- a/README.md +++ b/README.md @@ -1,26 +1,317 @@ # kubectl-application-shell -### Description +A kubectl plugin that creates safe debug pods from existing Kubernetes deployments. -This kubectl plugin allows you to solve this issue: +## Description -> You may want to run a one off sql query wrapped up in a script just to glean some information from the database. The traditional way to do this is by using `kubectl exec` into an existing pod running as part of your deployment. However, this may cause issues if your one-off command ends up killing the pod, which may impact production (dropping connections etc). It's much safer just to spin up a pod to do the one off task. +This kubectl plugin solves a common problem in Kubernetes debugging: -This command essentially wraps around `kubectl run`. It will allow you to specify a deployment name and the deployment namespace, and it will automatically grab the currently running image, resource limits/requests, config and secret mappings so that you have the required environment variables to complete your one-off task without impacting the existing deployment. +> When you need to run one-off tasks (SQL queries, file inspections, environment debugging), the traditional approach is `kubectl exec` into a running pod. However, this can cause issues if your command crashes the pod, potentially impacting production traffic. -### Pre-requisites +**kubectl-application-shell** provides a safer alternative by creating isolated debug pods that inherit the configuration from your existing deployment (image, environment variables, volumes, resource limits) without affecting running services. -- [kubectl](https://kubernetes.io/docs/tasks/tools/install-kubectl/) -- [jq](https://github.com/stedolan/jq/wiki/Installation) +## Features -### Installation +- 🚀 **Safe Debugging**: Creates isolated pods that don't impact production +- 🔄 **Auto-Configuration**: Inherits image, env vars, volumes, and resources from target deployment +- 🔗 **Session Management**: Resume disconnected sessions, list active sessions, manage multiple debugging sessions +- 🛡️ **Security Options**: Service accounts, user contexts, privileged mode, network policies +- 💾 **Storage Access**: Mount PVCs for persistent data debugging +- 📝 **Session Logging**: Record debugging sessions for later analysis +- ⚡ **Resource Control**: Override CPU/memory limits for specific debugging needs +- 🎯 **Node Targeting**: Schedule on specific nodes for hardware-specific debugging -This uses pipenv in order to manage dependencies. You need Python 3. +## Prerequisites -### Usage +- [kubectl](https://kubernetes.io/docs/tasks/tools/install-kubectl/) (automatically downloaded if needed) +- Python 3.9+ -Run `pipenv install` and `pipenv shell`. +## Installation -You can then run `kubeas` and it will give you the options. +### Using pipenv (Development) +```bash +pipenv install +pipenv shell +``` -It will automatically grab the image, resource limits/requests, config and secret mappings from the specified deployment. +### Using pip (Production) +```bash +pip install -e . +``` + +## Quick Start + +```bash +# 1. Create a debug session from a deployment +kubeas production api-server --run + +# 2. List active sessions (from another terminal) +kubeas list-sessions + +# 3. Resume the session later +kubeas resume-session debug-api-server-abc123 -n production --run + +# 4. Clean up when done +kubeas terminate-session debug-api-server-abc123 -n production +``` + +## Basic Usage + +```bash +# Create debug pod from deployment +kubeas + +# Run immediately without showing command +kubeas --run + +# Use custom shell/command +kubeas /bin/zsh + +# Pass arguments to the command +kubeas python3 script.py --verbose +``` + +## Advanced Options + +### Authentication & Authorization +```bash +# Use specific service account for RBAC debugging +kubeas --service-account debug-sa + +# Run as specific user ID +kubeas --run-as-user 1000 +``` + +### Environment & Runtime +```bash +# Override environment variables +kubeas --env DEBUG=true --env LOG_LEVEL=debug + +# Set working directory +kubeas --workdir /app/scripts + +# Custom timeout for pod startup +kubeas --timeout 10m +``` + +### Resource Management +```bash +# Override resource limits +kubeas --cpu-limit 2 --memory-limit 4Gi + +# Set specific resource requests +kubeas --cpu-request 500m --memory-request 1Gi +``` + +### Storage & Networking +```bash +# Mount persistent volume +kubeas --mount-pvc data-pvc + +# Use host network for network debugging +kubeas --host-network + +# Target specific nodes +kubeas --node-selector kubernetes.io/arch=amd64 +``` + +### Security & Debugging +```bash +# Run in privileged mode for system debugging +kubeas --privileged + +# Create network policies for isolation +kubeas --network-policy block-all-egress + +# Allow specific ports through network policy +kubeas --network-policy allow-port --network-port 8080 +``` + +### Session Management +```bash +# Log entire session to file +kubeas --session-log debug-session.log + +# Non-interactive mode for automation +kubeas --no-tty + +# List active debug sessions +kubeas list-sessions +kubeas list-sessions -n production + +# Resume an existing session +kubeas resume-session debug-api-abc123 -n production + +# Terminate a session +kubeas terminate-session debug-api-abc123 -n production +``` + +## Common Use Cases + +### Database Debugging +```bash +# Debug database connectivity with mounted data +kubeas production api-server --mount-pvc postgres-data --env DEBUG_DB=true +``` + +### Network Troubleshooting +```bash +# Debug network issues with host network access +kubeas production web-app --host-network --privileged +``` + +### Performance Analysis +```bash +# Run with higher resources for performance testing +kubeas staging api --cpu-limit 4 --memory-limit 8Gi --session-log perf-test.log +``` + +### RBAC Testing +```bash +# Test with different service account permissions +kubeas production worker --service-account restricted-sa --no-tty +``` + +### Session Management Workflow +```bash +# 1. Start a debug session (persists after disconnection) +kubeas production api-server + +# 2. List active sessions to see what's running +kubeas list-sessions + +# 3. Resume a session from another terminal/machine +kubeas resume-session debug-api-server-abc123 -n production --run + +# 4. Clean up when done +kubeas terminate-session debug-api-server-abc123 -n production +``` + +## How It Works + +1. **Deployment Inspection**: Queries the target deployment to extract configuration +2. **Kubectl Download**: Automatically downloads cluster-compatible kubectl binary +3. **Pod Generation**: Creates pod spec inheriting deployment configuration with session tracking labels +4. **Resource Creation**: Applies any network policies or additional resources +5. **Session Management**: Launches interactive shell with persistent session capability +6. **Session Tracking**: Uses pod labels with user identifiers for session ownership and management +7. **Cleanup**: NetworkPolicies auto-cleanup; pods persist for resumption until manually terminated + +## Advanced Features + +### Network Policies +When using `--network-policy`, the tool creates Kubernetes NetworkPolicy resources: +- `block-all-ingress`: Deny incoming traffic +- `block-all-egress`: Deny outgoing traffic +- `block-all`: Deny all traffic +- `allow-port`: Allow specific port (requires `--network-port`) +- `allow-port-range`: Allow port range (requires `--network-port-range`) + +### Session Management +- **Persistent Sessions**: Debug pods persist after terminal disconnection for later resumption +- **User Isolation**: Each user can only manage their own sessions (based on system username) +- **Cross-Platform**: Session state stored in Kubernetes labels, works from any machine +- **Session Discovery**: `list-sessions` shows all active debug sessions with metadata + +### Automatic Cleanup +- **NetworkPolicies**: Automatically deleted when sessions end +- **Debug Pods**: Persist for resumption; use `terminate-session` for manual cleanup +- **Cleanup Reminders**: Tool shows cleanup commands when not auto-running + +## Command Reference + +### Main Commands +```bash +kubeas [command] [options] # Create debug session +kubeas list-sessions [-n namespace] # List active sessions +kubeas resume-session -n [options] # Resume existing session +kubeas terminate-session -n # Terminate session +``` + +### Common Options +```bash +--run # Execute immediately +--context # Use specific kubeconfig context +--service-account # Use specific service account +--env KEY=VALUE # Set environment variables +--cpu-limit # Override CPU limit +--memory-limit # Override memory limit +--mount-pvc # Mount persistent volume +--session-log # Log session to file +--privileged # Run in privileged mode +--host-network # Use host network +--no-tty # Non-interactive mode +``` + +## Troubleshooting + +### Common Issues +- **Permission Denied**: Use `--service-account` with appropriate RBAC +- **Image Pull Errors**: Verify the deployment's image is accessible +- **Network Issues**: Try `--host-network` for network debugging +- **Resource Constraints**: Use custom `--cpu-limit` and `--memory-limit` +- **Session Not Found**: Check namespace with `kubeas list-sessions` +- **Cannot Resume Session**: Verify you own the session (sessions are user-isolated) + +### Session Management Issues +```bash +# Check if sessions exist +kubeas list-sessions + +# Verify session ownership (only your username sessions are manageable) +kubeas list-sessions -n + +# Force cleanup if needed (requires kubectl access) +kubectl delete pod -n +``` + +### Getting Help +```bash +kubeas --help # Main command help +kubeas list-sessions --help # Session listing help +kubeas resume-session --help # Resume command help +kubeas terminate-session --help # Terminate command help +``` + +## Contributing + +### Development Setup +```bash +git clone +cd kubectl-application-shell +pipenv install --dev +pipenv shell +``` + +### Testing +```bash +# Test against a real cluster +kubeas test-namespace test-deployment --run + +# Test session management +kubeas list-sessions +``` + +### Architecture +This project uses: +- **Typer** for CLI interface with multiple commands +- **Rich** for colored output and tables +- **Kubernetes Python Client** for API interactions +- **Pipenv** for dependency management +- **Pod Labels** for session tracking and user isolation + +### Key Files +- `src/kubectl_application_shell/app.py` - Main debug session creation logic +- `src/kubectl_application_shell/cli.py` - CLI commands and interface +- `src/kubectl_application_shell/func.py` - Kubernetes API functions and session management +- `src/kubectl_application_shell/console.py` - Rich console instance + +### Adding Features +1. Add new CLI options to `app.py` main function +2. Update pod spec generation logic +3. Add corresponding functions to `func.py` if needed +4. Update documentation in README.md and CLAUDE.md + +## License + +See project license file. diff --git a/pyproject.toml b/pyproject.toml index 0c9937c..055e20b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ dependencies = [ ] [project.scripts] -kubeas = "kubectl_application_shell.cli:app" +kubeas = "kubectl_application_shell:run_cli" [build-system] requires = ["hatchling"] diff --git a/src/kubectl_application_shell/__init__.py b/src/kubectl_application_shell/__init__.py index e69de29..625fb06 100644 --- a/src/kubectl_application_shell/__init__.py +++ b/src/kubectl_application_shell/__init__.py @@ -0,0 +1,20 @@ +"""kubectl-application-shell package""" + +import sys +from typing import List + + +def run_cli(): + """Entry point that handles default command behavior""" + from .cli import app + + # If we have arguments and the first one is not a known command + known_commands = ["main", "list-sessions", "resume-session", "terminate-session", "--help", "-h"] + + if len(sys.argv) > 1 and sys.argv[1] not in known_commands: + # Check if the first argument starts with -- (it's an option) + if not sys.argv[1].startswith("-"): + # Insert "main" as the command + sys.argv.insert(1, "main") + + app(prog_name="kubeas") \ No newline at end of file diff --git a/src/kubectl_application_shell/__main__.py b/src/kubectl_application_shell/__main__.py index 4b03e5d..ab53957 100644 --- a/src/kubectl_application_shell/__main__.py +++ b/src/kubectl_application_shell/__main__.py @@ -1,5 +1,5 @@ """Helper module to run the CLI application.""" if __name__ == "__main__": - from kubectl_application_shell.cli import app - app(prog_name="kubeas") + from kubectl_application_shell import run_cli + run_cli() diff --git a/src/kubectl_application_shell/app.py b/src/kubectl_application_shell/app.py index b6f6f82..135cc93 100644 --- a/src/kubectl_application_shell/app.py +++ b/src/kubectl_application_shell/app.py @@ -9,7 +9,11 @@ import typer from .console import console -from .func import get_deployment_info, get_kubectl, get_kube_version +from .func import ( + get_deployment_info, get_kubectl, get_kube_version, create_network_policy, + delete_network_policy, get_user_identifier, list_debug_sessions, + get_debug_session, terminate_debug_session, resume_debug_session +) def main( @@ -37,6 +41,78 @@ def main( List[str], typer.Argument(help="The arguments to pass to the command/entrypoint/shell."), ] = None, + cpu_request: Annotated[ + Optional[str], + typer.Option("--cpu-request", help="CPU request for the debug pod (e.g., '100m', '0.5').", show_default=False), + ] = None, + cpu_limit: Annotated[ + Optional[str], + typer.Option("--cpu-limit", help="CPU limit for the debug pod (e.g., '500m', '1').", show_default=False), + ] = None, + memory_request: Annotated[ + Optional[str], + typer.Option("--memory-request", help="Memory request for the debug pod (e.g., '256Mi', '1Gi').", show_default=False), + ] = None, + memory_limit: Annotated[ + Optional[str], + typer.Option("--memory-limit", help="Memory limit for the debug pod (e.g., '512Mi', '2Gi').", show_default=False), + ] = None, + network_policy: Annotated[ + Optional[str], + typer.Option("--network-policy", help="NetworkPolicy type: block-all-ingress, block-all-egress, block-all, allow-port, allow-port-range", show_default=False), + ] = None, + network_port: Annotated[ + Optional[str], + typer.Option("--network-port", help="Port number to allow (used with allow-port policy)", show_default=False), + ] = None, + network_port_range: Annotated[ + Optional[str], + typer.Option("--network-port-range", help="Port range to allow, e.g. '8000-8080' (used with allow-port-range policy)", show_default=False), + ] = None, + session_log: Annotated[ + Optional[str], + typer.Option("--session-log", help="File path to log the session output.", show_default=False), + ] = None, + service_account: Annotated[ + Optional[str], + typer.Option("--service-account", help="Service account to use for RBAC permissions.", show_default=False), + ] = None, + env_vars: Annotated[ + Optional[List[str]], + typer.Option("--env", help="Environment variables to set (format: KEY=VALUE). Can be used multiple times.", show_default=False), + ] = None, + working_directory: Annotated[ + Optional[str], + typer.Option("--workdir", help="Working directory for the debug pod.", show_default=False), + ] = None, + node_selector: Annotated[ + Optional[str], + typer.Option("--node-selector", help="Node selector constraint (format: KEY=VALUE).", show_default=False), + ] = None, + timeout: Annotated[ + Optional[str], + typer.Option("--timeout", help="Pod running timeout (default: 5m).", show_default=False), + ] = None, + mount_pvc: Annotated[ + Optional[str], + typer.Option("--mount-pvc", help="PVC to mount at /mnt/data.", show_default=False), + ] = None, + no_tty: Annotated[ + bool, + typer.Option("--no-tty", help="Disable TTY allocation (non-interactive mode)."), + ] = False, + privileged: Annotated[ + bool, + typer.Option("--privileged", help="Run in privileged mode for system debugging."), + ] = False, + host_network: Annotated[ + bool, + typer.Option("--host-network", help="Use host network for network debugging."), + ] = False, + run_as_user: Annotated[ + Optional[int], + typer.Option("--run-as-user", help="User ID to run as.", show_default=False), + ] = None, run: Annotated[ bool, typer.Option("--run", help="Run the debug pod."), @@ -72,46 +148,161 @@ def main( container_name = deployment_info["spec"]["template"]["spec"]["containers"][0]["name"] image = image or deployment_info["spec"]["template"]["spec"]["containers"][0]["image"] - env = deployment_info["spec"]["template"]["spec"]["containers"][0].get("env", None) + env = deployment_info["spec"]["template"]["spec"]["containers"][0].get("env", []) env_from = deployment_info["spec"]["template"]["spec"]["containers"][0].get("envFrom", None) - resources = deployment_info["spec"]["template"]["spec"]["containers"][0].get("resources", None) + + # Add custom environment variables + if env_vars: + for env_var in env_vars: + if "=" in env_var: + key, value = env_var.split("=", 1) + env.append({"name": key, "value": value}) + else: + console.print(f":warning: Skipping invalid env var format: {env_var}") + env = env if env else None + + # Build custom resources if specified, otherwise use deployment resources + resources = deployment_info["spec"]["template"]["spec"]["containers"][0].get("resources", {}) + if any([cpu_request, cpu_limit, memory_request, memory_limit]): + custom_resources = {} + if cpu_request or memory_request: + custom_resources["requests"] = {} + if cpu_request: + custom_resources["requests"]["cpu"] = cpu_request + if memory_request: + custom_resources["requests"]["memory"] = memory_request + if cpu_limit or memory_limit: + custom_resources["limits"] = {} + if cpu_limit: + custom_resources["limits"]["cpu"] = cpu_limit + if memory_limit: + custom_resources["limits"]["memory"] = memory_limit + resources = custom_resources + volume_mounts = deployment_info["spec"]["template"]["spec"]["containers"][0].get( - "volumeMounts", None + "volumeMounts", [] ) - volumes = deployment_info["spec"]["template"]["spec"].get("volumes", None) - kubectl_overrides = json.dumps( - { - "spec": { - "containers": [ - { - "name": container_name, - "image": image, - "command": [command], - "args": args if args else [], - "env": env, - "envFrom": env_from, - "resources": resources, - "stdin": True, - "stdinOnce": True, - "tty": True, - "volumeMounts": volume_mounts, - } - ], - "volumes": volumes, - }, + volumes = deployment_info["spec"]["template"]["spec"].get("volumes", []) + + # Add PVC mount if specified + if mount_pvc: + pvc_volume = { + "name": "debug-pvc", + "persistentVolumeClaim": { + "claimName": mount_pvc + } } - ) + pvc_mount = { + "name": "debug-pvc", + "mountPath": "/mnt/data" + } + volumes.append(pvc_volume) + volume_mounts.append(pvc_mount) + + # Convert back to None if empty for clean JSON + volume_mounts = volume_mounts if volume_mounts else None + volumes = volumes if volumes else None + # Build container spec + container_spec = { + "name": container_name, + "image": image, + "command": [command], + "args": args if args else [], + "env": env, + "envFrom": env_from, + "resources": resources, + "stdin": True, + "stdinOnce": True, + "tty": not no_tty, + "volumeMounts": volume_mounts, + } + + # Add working directory if specified + if working_directory: + container_spec["workingDir"] = working_directory + + # Add security context if specified + security_context = {} + if run_as_user is not None: + security_context["runAsUser"] = run_as_user + if privileged: + security_context["privileged"] = True + if security_context: + container_spec["securityContext"] = security_context + + # Build pod spec + pod_spec = { + "spec": { + "containers": [container_spec], + "volumes": volumes, + }, + } + + # Add service account if specified + if service_account: + pod_spec["spec"]["serviceAccountName"] = service_account + + # Add node selector if specified + if node_selector and "=" in node_selector: + key, value = node_selector.split("=", 1) + pod_spec["spec"]["nodeSelector"] = {key: value} + + # Add host network if specified + if host_network: + pod_spec["spec"]["hostNetwork"] = True + + # Add session tracking labels + labels = { + "app": "kubectl-application-shell", + "debug-user": get_user_identifier(), + "debug-deployment": deployment, + } + + # Add NetworkPolicy targeting label if needed + if network_policy: + labels["debug-pod"] = pod_name + + pod_spec["metadata"] = {"labels": labels} + + kubectl_overrides = json.dumps(pod_spec) name_random = "".join(random.choices(string.ascii_lowercase + string.digits, k=5)) + pod_name = f"debug-{deployment}-{name_random}" - cmd = "".join( - [ - f"{kubectl} run -it --rm --restart=Never --namespace={namespace} ", - f"--context={context} " if context else " ", - f"--image={image} --pod-running-timeout=5m debug-{deployment}-{name_random} ", - f"--overrides='{kubectl_overrides}'", - ] - ) + # Create NetworkPolicy if specified + policy_name = None + if network_policy: + policy_name = create_network_policy( + namespace=namespace, + pod_name=pod_name, + policy_type=network_policy, + port=network_port, + port_range=network_port_range, + context=context + ) + if not policy_name: + console.print(":fire: Failed to create NetworkPolicy, continuing without it") + + # Build the base kubectl command + interactive_flags = "-it" if not no_tty else "-i" + pod_timeout = timeout or "5m" + + cmd_parts = [ + f"{kubectl} run {interactive_flags} --restart=Never --namespace={namespace} ", + f"--context={context} " if context else " ", + f"--image={image} --pod-running-timeout={pod_timeout} {pod_name} ", + f"--overrides='{kubectl_overrides}'", + ] + + base_cmd = "".join(cmd_parts) + + # Add session logging if specified + if session_log: + # Use script command to log the session + cmd = f"script -q {session_log} -c \"{base_cmd}\"" + console.print(f":memo: Session will be logged to [bold green]{session_log}[/bold green]") + else: + cmd = base_cmd if not run: # Return the kubectl command for them to run. @@ -120,8 +311,29 @@ def main( "(or add `--run` to run it automatically):" ) print(cmd) + if policy_name: + console.print( + f":warning: [bold yellow]Don't forget to clean up the NetworkPolicy:[/bold yellow] " + f"kubectl delete networkpolicy {policy_name} -n {namespace}" + ) + console.print( + f":information: Session will be named [bold green]{pod_name}[/bold green]. " + f"Use [bold blue]kubeas list-sessions[/bold blue] to see active sessions." + ) raise typer.Exit() # Run the kubectl command. - console.print(":rocket: Running the debug pod!") - os.system(cmd) + console.print(f":rocket: Running debug session [bold green]{pod_name}[/bold green]!") + if session_log: + console.print(f":memo: Logging session to {session_log}") + + try: + exit_code = os.system(cmd) + finally: + # Clean up NetworkPolicy if it was created + if policy_name: + console.print(":broom: Cleaning up NetworkPolicy...") + delete_network_policy(namespace, policy_name, context) + + # Note about session cleanup + console.print(f":information: Session [bold yellow]{pod_name}[/bold yellow] is still running. Use [bold blue]kubeas terminate-session {pod_name} -n {namespace}[/bold blue] to clean up.") diff --git a/src/kubectl_application_shell/cli.py b/src/kubectl_application_shell/cli.py index 4f3e2cf..956e6b6 100644 --- a/src/kubectl_application_shell/cli.py +++ b/src/kubectl_application_shell/cli.py @@ -1,11 +1,281 @@ """CLI entry point for the kubectl_application_shell package.""" +import os +from typing import Annotated, Optional, List + import typer +from rich.table import Table -from .app import main +from .console import console +from .func import ( + get_kubectl, get_kube_version, list_debug_sessions, + terminate_debug_session, resume_debug_session +) app = typer.Typer(add_completion=False) -app.command()(main) + + +# Main command for creating debug sessions +@app.command(name="main", help="produce a debug pod for a Kubernetes deployment") +def main_command( + namespace: Annotated[ + str, + typer.Argument(help="The namespace of the deployment."), + ], + deployment: Annotated[ + str, + typer.Argument(help="The deployment to debug."), + ], + context: Annotated[ + Optional[str], + typer.Option(help="The kubeconfig context to use.", show_default=False), + ] = None, + image: Annotated[ + Optional[str], + typer.Option(help="The image to run in the debug pod.", show_default=False), + ] = None, + command: Annotated[ + str, + typer.Argument(help="The command/entrypoint/shell to run in the debug pod."), + ] = "/bin/bash", + args: Annotated[ + List[str], + typer.Argument(help="The arguments to pass to the command/entrypoint/shell."), + ] = None, + cpu_request: Annotated[ + Optional[str], + typer.Option("--cpu-request", help="CPU request for the debug pod (e.g., '100m', '0.5').", show_default=False), + ] = None, + cpu_limit: Annotated[ + Optional[str], + typer.Option("--cpu-limit", help="CPU limit for the debug pod (e.g., '500m', '1').", show_default=False), + ] = None, + memory_request: Annotated[ + Optional[str], + typer.Option("--memory-request", help="Memory request for the debug pod (e.g., '256Mi', '1Gi').", show_default=False), + ] = None, + memory_limit: Annotated[ + Optional[str], + typer.Option("--memory-limit", help="Memory limit for the debug pod (e.g., '512Mi', '2Gi').", show_default=False), + ] = None, + network_policy: Annotated[ + Optional[str], + typer.Option("--network-policy", help="NetworkPolicy type: block-all-ingress, block-all-egress, block-all, allow-port, allow-port-range", show_default=False), + ] = None, + network_port: Annotated[ + Optional[str], + typer.Option("--network-port", help="Port number to allow (used with allow-port policy)", show_default=False), + ] = None, + network_port_range: Annotated[ + Optional[str], + typer.Option("--network-port-range", help="Port range to allow, e.g. '8000-8080' (used with allow-port-range policy)", show_default=False), + ] = None, + session_log: Annotated[ + Optional[str], + typer.Option("--session-log", help="File path to log the session output.", show_default=False), + ] = None, + service_account: Annotated[ + Optional[str], + typer.Option("--service-account", help="Service account to use for RBAC permissions.", show_default=False), + ] = None, + env_vars: Annotated[ + Optional[List[str]], + typer.Option("--env", help="Environment variables to set (format: KEY=VALUE). Can be used multiple times.", show_default=False), + ] = None, + working_directory: Annotated[ + Optional[str], + typer.Option("--workdir", help="Working directory for the debug pod.", show_default=False), + ] = None, + node_selector: Annotated[ + Optional[str], + typer.Option("--node-selector", help="Node selector constraint (format: KEY=VALUE).", show_default=False), + ] = None, + timeout: Annotated[ + Optional[str], + typer.Option("--timeout", help="Pod running timeout (default: 5m).", show_default=False), + ] = None, + mount_pvc: Annotated[ + Optional[str], + typer.Option("--mount-pvc", help="PVC to mount at /mnt/data.", show_default=False), + ] = None, + no_tty: Annotated[ + bool, + typer.Option("--no-tty", help="Disable TTY allocation (non-interactive mode)."), + ] = False, + privileged: Annotated[ + bool, + typer.Option("--privileged", help="Run in privileged mode for system debugging."), + ] = False, + host_network: Annotated[ + bool, + typer.Option("--host-network", help="Use host network for network debugging."), + ] = False, + run_as_user: Annotated[ + Optional[int], + typer.Option("--run-as-user", help="User ID to run as.", show_default=False), + ] = None, + run: Annotated[ + bool, + typer.Option("--run", help="Run the debug pod."), + ] = False, +) -> None: + """produce a debug pod for a Kubernetes deployment""" + from .app import main + # Call the actual main function with all parameters + main( + namespace=namespace, + deployment=deployment, + context=context, + image=image, + command=command, + args=args, + cpu_request=cpu_request, + cpu_limit=cpu_limit, + memory_request=memory_request, + memory_limit=memory_limit, + network_policy=network_policy, + network_port=network_port, + network_port_range=network_port_range, + session_log=session_log, + service_account=service_account, + env_vars=env_vars, + working_directory=working_directory, + node_selector=node_selector, + timeout=timeout, + mount_pvc=mount_pvc, + no_tty=no_tty, + privileged=privileged, + host_network=host_network, + run_as_user=run_as_user, + run=run + ) + + +@app.command("list-sessions") +def list_sessions( + namespace: Annotated[ + Optional[str], + typer.Option("-n", "--namespace", help="Namespace to list sessions from (all namespaces if not specified)"), + ] = None, + context: Annotated[ + Optional[str], + typer.Option(help="The kubeconfig context to use.", show_default=False), + ] = None, +) -> None: + """List active debug sessions""" + + sessions = list_debug_sessions(namespace, context) + + if not sessions: + if namespace: + console.print(f":information: No active debug sessions found in namespace [bold]{namespace}[/bold]") + else: + console.print(":information: No active debug sessions found") + return + + # Create table for display + table = Table(title="Active Debug Sessions") + table.add_column("Session Name", style="cyan") + table.add_column("Namespace", style="magenta") + table.add_column("User", style="green") + table.add_column("Deployment", style="blue") + table.add_column("Status", style="yellow") + table.add_column("Created", style="dim") + table.add_column("Node", style="dim") + + for session in sessions: + table.add_row( + session["name"], + session["namespace"], + session["user"], + session["deployment"], + session["status"], + session["created"], + session["node"] + ) + + console.print(table) + + +@app.command("resume-session") +def resume_session_cmd( + session_name: Annotated[str, typer.Argument(help="Name of the debug session to resume")], + namespace: Annotated[ + str, + typer.Option("-n", "--namespace", help="Namespace of the session"), + ], + context: Annotated[ + Optional[str], + typer.Option(help="The kubeconfig context to use.", show_default=False), + ] = None, + session_log: Annotated[ + Optional[str], + typer.Option("--session-log", help="File path to log the session output.", show_default=False), + ] = None, + no_tty: Annotated[ + bool, + typer.Option("--no-tty", help="Disable TTY allocation (non-interactive mode)."), + ] = False, + run: Annotated[ + bool, + typer.Option("--run", help="Execute immediately instead of showing the command."), + ] = False, +) -> None: + """Resume an existing debug session""" + + # Get kubectl binary + kube_version = get_kube_version(context) + if not kube_version: + raise typer.Exit(code=1) + + kubectl = get_kubectl(kube_version) + + # Generate resume command + cmd = resume_debug_session( + session_name=session_name, + namespace=namespace, + kubectl_path=kubectl, + context=context, + session_log=session_log, + no_tty=no_tty + ) + + if not cmd: + raise typer.Exit(code=1) + + if not run: + console.print( + f":arrows_counterclockwise: Ready to resume session [bold green]{session_name}[/bold green]! " + "[bold blue]Run this command:[/bold blue]" + ) + print(cmd) + return + + # Execute the resume command + console.print(f":arrows_counterclockwise: Resuming session [bold green]{session_name}[/bold green]...") + if session_log: + console.print(f":memo: Logging session to {session_log}") + + os.system(cmd) + + +@app.command("terminate-session") +def terminate_session_cmd( + session_name: Annotated[str, typer.Argument(help="Name of the debug session to terminate")], + namespace: Annotated[ + str, + typer.Option("-n", "--namespace", help="Namespace of the session"), + ], + context: Annotated[ + Optional[str], + typer.Option(help="The kubeconfig context to use.", show_default=False), + ] = None, +) -> None: + """Terminate a debug session""" + + success = terminate_debug_session(session_name, namespace, context) + if not success: + raise typer.Exit(code=1) if __name__ == "__main__": diff --git a/src/kubectl_application_shell/func.py b/src/kubectl_application_shell/func.py index 2cdb801..70ff8f5 100644 --- a/src/kubectl_application_shell/func.py +++ b/src/kubectl_application_shell/func.py @@ -5,7 +5,8 @@ import sys from pathlib import Path from shutil import which -from typing import Optional +from typing import Optional, List, Dict +import getpass import requests from kubernetes import client, config @@ -111,3 +112,305 @@ def get_deployment_info( return None return json.loads(deployment_info.data) + + +def create_network_policy( + namespace: str, + pod_name: str, + policy_type: str, + port: str = None, + port_range: str = None, + context: str = None, +) -> Optional[str]: + """create a NetworkPolicy for the debug pod""" + + network_v1 = client.NetworkingV1Api(get_api_client(context)) + + # Base policy name + policy_name = f"debug-{pod_name}-netpol" + + # Base pod selector + pod_selector = client.V1LabelSelector( + match_labels={"debug-pod": pod_name} + ) + + # Define policy templates + if policy_type == "block-all-ingress": + policy = client.V1NetworkPolicy( + metadata=client.V1ObjectMeta( + name=policy_name, + namespace=namespace + ), + spec=client.V1NetworkPolicySpec( + pod_selector=pod_selector, + policy_types=["Ingress"], + ingress=[] # Empty ingress rules = deny all + ) + ) + + elif policy_type == "block-all-egress": + policy = client.V1NetworkPolicy( + metadata=client.V1ObjectMeta( + name=policy_name, + namespace=namespace + ), + spec=client.V1NetworkPolicySpec( + pod_selector=pod_selector, + policy_types=["Egress"], + egress=[] # Empty egress rules = deny all + ) + ) + + elif policy_type == "block-all": + policy = client.V1NetworkPolicy( + metadata=client.V1ObjectMeta( + name=policy_name, + namespace=namespace + ), + spec=client.V1NetworkPolicySpec( + pod_selector=pod_selector, + policy_types=["Ingress", "Egress"], + ingress=[], + egress=[] + ) + ) + + elif policy_type == "allow-port" and port: + # Allow specific port ingress + port_rule = client.V1NetworkPolicyPort( + port=int(port), + protocol="TCP" + ) + ingress_rule = client.V1NetworkPolicyIngressRule( + ports=[port_rule] + ) + policy = client.V1NetworkPolicy( + metadata=client.V1ObjectMeta( + name=policy_name, + namespace=namespace + ), + spec=client.V1NetworkPolicySpec( + pod_selector=pod_selector, + policy_types=["Ingress"], + ingress=[ingress_rule] + ) + ) + + elif policy_type == "allow-port-range" and port_range: + # Parse port range (e.g., "8000-8080") + start_port, end_port = map(int, port_range.split("-")) + port_rule = client.V1NetworkPolicyPort( + port=start_port, + end_port=end_port, + protocol="TCP" + ) + ingress_rule = client.V1NetworkPolicyIngressRule( + ports=[port_rule] + ) + policy = client.V1NetworkPolicy( + metadata=client.V1ObjectMeta( + name=policy_name, + namespace=namespace + ), + spec=client.V1NetworkPolicySpec( + pod_selector=pod_selector, + policy_types=["Ingress"], + ingress=[ingress_rule] + ) + ) + + else: + console.print(f":fire: Unknown network policy type: {policy_type}") + return None + + try: + network_v1.create_namespaced_network_policy( + namespace=namespace, + body=policy + ) + console.print(f":shield: Created NetworkPolicy: [bold green]{policy_name}[/bold green]") + return policy_name + except ApiException as e: + console.print(f":fire: Failed to create NetworkPolicy: {e.reason}") + return None + + +def delete_network_policy( + namespace: str, + policy_name: str, + context: str = None, +) -> bool: + """delete a NetworkPolicy""" + + network_v1 = client.NetworkingV1Api(get_api_client(context)) + + try: + network_v1.delete_namespaced_network_policy( + name=policy_name, + namespace=namespace + ) + console.print(f":shield: Deleted NetworkPolicy: [bold red]{policy_name}[/bold red]") + return True + except ApiException as e: + console.print(f":fire: Failed to delete NetworkPolicy: {e.reason}") + return False + + +def get_user_identifier() -> str: + """get current user identifier for session tracking""" + return getpass.getuser() + + +def list_debug_sessions( + namespace: str = None, + context: str = None, +) -> List[Dict]: + """list active debug sessions (pods with debug labels)""" + + v1 = client.CoreV1Api(get_api_client(context)) + + try: + if namespace: + pods = v1.list_namespaced_pod( + namespace=namespace, + label_selector="app=kubectl-application-shell" + ) + else: + pods = v1.list_pod_for_all_namespaces( + label_selector="app=kubectl-application-shell" + ) + except (ApiException, MaxRetryError) as e: + console.print(f":fire: Unable to list debug sessions: {e.reason}") + return [] + + sessions = [] + for pod in pods.items: + labels = pod.metadata.labels or {} + session_info = { + "name": pod.metadata.name, + "namespace": pod.metadata.namespace, + "status": pod.status.phase, + "user": labels.get("debug-user", "unknown"), + "deployment": labels.get("debug-deployment", "unknown"), + "created": pod.metadata.creation_timestamp.strftime("%Y-%m-%d %H:%M:%S") if pod.metadata.creation_timestamp else "unknown", + "node": pod.spec.node_name or "pending" + } + sessions.append(session_info) + + return sessions + + +def get_debug_session( + session_name: str, + namespace: str, + context: str = None, +) -> Optional[Dict]: + """get details for a specific debug session""" + + v1 = client.CoreV1Api(get_api_client(context)) + + try: + pod = v1.read_namespaced_pod( + name=session_name, + namespace=namespace + ) + except (ApiException, MaxRetryError) as e: + console.print(f":fire: Unable to get session {session_name}: {e.reason}") + return None + + # Verify it's a debug session + labels = pod.metadata.labels or {} + if labels.get("app") != "kubectl-application-shell": + console.print(f":fire: Pod {session_name} is not a debug session") + return None + + return { + "name": pod.metadata.name, + "namespace": pod.metadata.namespace, + "status": pod.status.phase, + "user": labels.get("debug-user", "unknown"), + "deployment": labels.get("debug-deployment", "unknown"), + "created": pod.metadata.creation_timestamp.strftime("%Y-%m-%d %H:%M:%S") if pod.metadata.creation_timestamp else "unknown", + "node": pod.spec.node_name or "pending", + "container_name": pod.spec.containers[0].name if pod.spec.containers else "unknown" + } + + +def terminate_debug_session( + session_name: str, + namespace: str, + context: str = None, +) -> bool: + """terminate a specific debug session""" + + v1 = client.CoreV1Api(get_api_client(context)) + + # First verify it's a debug session + session = get_debug_session(session_name, namespace, context) + if not session: + return False + + # Check if current user owns the session + current_user = get_user_identifier() + if session["user"] != current_user: + console.print(f":fire: Cannot terminate session owned by {session['user']} (you are {current_user})") + return False + + try: + v1.delete_namespaced_pod( + name=session_name, + namespace=namespace + ) + console.print(f":skull: Terminated debug session: [bold red]{session_name}[/bold red]") + return True + except (ApiException, MaxRetryError) as e: + console.print(f":fire: Failed to terminate session: {e.reason}") + return False + + +def resume_debug_session( + session_name: str, + namespace: str, + kubectl_path: Path, + context: str = None, + session_log: str = None, + no_tty: bool = False, +) -> str: + """generate command to resume a debug session""" + + # Verify session exists and get details + session = get_debug_session(session_name, namespace, context) + if not session: + return None + + if session["status"] != "Running": + console.print(f":fire: Session {session_name} is not running (status: {session['status']})") + return None + + # Check if current user owns the session + current_user = get_user_identifier() + if session["user"] != current_user: + console.print(f":fire: Cannot resume session owned by {session['user']} (you are {current_user})") + return None + + # Build kubectl exec command + interactive_flags = "-it" if not no_tty else "-i" + container_name = session["container_name"] + + cmd_parts = [ + f"{kubectl_path} exec {interactive_flags} --namespace={namespace} ", + f"--context={context} " if context else " ", + f"{session_name} ", + f"--container={container_name} -- /bin/bash" + ] + + base_cmd = "".join(cmd_parts) + + # Add session logging if specified + if session_log: + cmd = f"script -q {session_log} -c \"{base_cmd}\"" + console.print(f":memo: Session will be logged to [bold green]{session_log}[/bold green]") + else: + cmd = base_cmd + + return cmd