Bake in Kubernetes port-forwarding

AGENTS.md

@@ -25,6 +25,7 @@ A command line client for MySQL with auto-completion and syntax highlighting.
 ├── mycli/config.py                       # configuration file readers and utilities
 ├── mycli/constants.py                    # shared constants
 ├── mycli/key_bindings.py                 # prompt_toolkit key binding utilities
+├── mycli/kubectl_tunnel.py               # connection over kubectl tunnel
 ├── mycli/lexer.py                        # extends `MySqlLexer` from Pygments
 ├── mycli/main.py                         # processes CLI arguments
 ├── mycli/main_modes/                     # main execution paths

README.md

@@ -85,6 +85,7 @@ mycli --help
 * [Polars](https://pola.rs) dataframe [transforms and plots](https://github.com/dbcli/mycli/blob/main/doc/transforms.md) with `.|`, and Parquet saves with `.>`.
 * Support for [querying LLMs](https://www.mycli.net/llm) with context derived from your schema using `/llm`.
 * Support for storing passwords in the system keyring.
+* Integrations with: [SSH](https://www.openssh.org/), [Kubernetes](https://kubernetes.io/), [Vault](https://github.com/hashicorp/vault), and [Boundary](https://github.com/hashicorp/boundary).
 
 Mycli creates a config file `~/.myclirc` on the first run; you can use the
 options in that file to configure the above features, and more.

changelog.md

@@ -4,6 +4,7 @@ Upcoming (TBD)
 Features
 --------
 * Add completion on leading-abbreviated pathname elements.
+* Add port-forwarding into a Kubernetes cluster with `--kubectl-resource`.
 
 
 2.21.1 (2026/09/07)

mycli/TIPS

@@ -42,6 +42,8 @@ the --login-path option lets you work with login-path files!
 
 connect via an SSH tunnel with --ssh-jump!
 
+connect to a Kubernetes pod with --kubectl-resource!
+
 ###
 ### commands
 ###

mycli/cli_runner.py

@@ -359,6 +359,8 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non
             mycli.prompt_format = cli_args.prompt or params[0] or mycli.prompt_format
         if params := dsn_params.get('ssh_jump'):
             cli_args.ssh_jump = cli_args.ssh_jump or params[0]
+        if params := dsn_params.get('kubectl_resource'):
+            cli_args.kubectl_resource = cli_args.kubectl_resource or params[0]
         if params := dsn_params.get('boundary_id'):
             cli_args.boundary_id = cli_args.boundary_id or params[0]
         if params := dsn_params.get('vault_address'):
@@ -372,8 +374,12 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non
         if params := dsn_params.get('vault_username_field'):
             cli_args.vault_username_field = cli_args.vault_username_field or params[0]
 
-    if cli_args.ssh_jump and cli_args.boundary_id:
-        click.secho('Error: --ssh-jump and --boundary-id are incompatible.', err=True, fg='red')
+    tunnel_options = [cli_args.ssh_jump, cli_args.kubectl_resource, cli_args.boundary_id]
+    if sum(option is not None and bool(option) for option in tunnel_options) > 1:
+        click.secho('Error: --ssh-jump, --kubectl-resource, and --boundary-id are mutually exclusive.', err=True, fg='red')
+        sys.exit(1)
+    if cli_args.kubectl_options and not cli_args.kubectl_resource:
+        click.secho('Error: --kubectl-options requires --kubectl-resource.', err=True, fg='red')
         sys.exit(1)
 
     keepalive_ticks = cli_args.keepalive_ticks if cli_args.keepalive_ticks is not None else mycli.default_keepalive_ticks
@@ -521,6 +527,8 @@ def load_vault_password() -> str | None:
             keepalive_ticks=keepalive_ticks,
             ssh_jump=cli_args.ssh_jump,
             ssh_cli_options=cli_args.ssh_options,
+            kubectl_resource=cli_args.kubectl_resource,
+            kubectl_cli_options=cli_args.kubectl_options,
             vault_address=cli_args.vault_address,
             vault_mount=cli_args.vault_mount,
             vault_secret=cli_args.vault_secret,

mycli/client.py

@@ -34,6 +34,7 @@
     write_default_config,
 )
 from mycli.constants import DEFAULT_PROMPT
+from mycli.kubectl_tunnel import KubectlTunnel
 from mycli.main_modes import repl as repl_package
 from mycli.output import OutputMixin
 from mycli.packages import special
@@ -80,6 +81,7 @@ def __init__(
     ) -> None:
         self.sqlexecute = sqlexecute
         self.ssh_tunnel: SshTunnel | None = None
+        self.kubectl_tunnel: KubectlTunnel | None = None
         self.boundary_tunnel: BoundaryTunnel | None = None
         self.logfile = logfile
         self.login_path = login_path
@@ -260,6 +262,11 @@ def close(self) -> None:
                 self.ssh_tunnel.close()
             except Exception:
                 pass
+        if self.kubectl_tunnel is not None:
+            try:
+                self.kubectl_tunnel.close()
+            except Exception:
+                pass
         if self.boundary_tunnel:
             try:
                 self.boundary_tunnel.close()

mycli/client_connection.py

@@ -22,6 +22,7 @@
     EMPTY_PASSWORD_FLAG_SENTINEL,
     ER_MUST_CHANGE_PASSWORD_LOGIN,
 )
+from mycli.kubectl_tunnel import KubectlTunnel, KubectlTunnelError
 from mycli.packages.filepaths import guess_socket_location
 from mycli.packages.special.utils import format_connection_dsn
 from mycli.password_sources import PasswordCandidates
@@ -45,6 +46,7 @@ class ClientConnectionMixin:
         sqlexecute: Any
         logger: Any
         boundary_tunnel: BoundaryTunnel | None
+        kubectl_tunnel: KubectlTunnel | None
 
         def read_mylogin_cnf(self, cnf: Any) -> dict[str, Any]: ...
         def echo(self, *args: Any, **kwargs: Any) -> None: ...
@@ -67,6 +69,8 @@ def connect(
         keepalive_ticks: int | None = None,
         ssh_jump: str | None = None,
         ssh_cli_options: str | None = None,
+        kubectl_resource: str | None = None,
+        kubectl_cli_options: str | None = None,
         vault_address: str | None = None,
         vault_mount: str | None = None,
         vault_secret: str | None = None,
@@ -84,6 +88,7 @@ def connect(
         user_connection_config = self.config_without_package_defaults.get('connection', {})
         self.keepalive_ticks = keepalive_ticks
         self.ssh_tunnel = None
+        self.kubectl_tunnel = None
         self.selected_password = None
         self.boundary_tunnel = None
 
@@ -130,6 +135,28 @@ def connect(
                 except Exception:
                     pass
                 sys.exit(1)
+        elif kubectl_resource:
+            use_keyring = False
+            kubectl_executable = self.config.get('kubectl', {}).get('kubectl_executable', 'kubectl') or 'kubectl'
+            kubectl_config_options = self.config.get('kubectl', {}).get('kubectl_options') or None
+            try:
+                self.kubectl_tunnel = KubectlTunnel(
+                    resource=kubectl_resource,
+                    remote_port=int(int_port),
+                    kubectl_executable=kubectl_executable,
+                    kubectl_config_options=kubectl_config_options,
+                    kubectl_cli_options=kubectl_cli_options,
+                )
+                self.kubectl_tunnel.start()
+                socket = None
+            except (OSError, ValueError, KubectlTunnelError) as exc:
+                click.secho(f'Error: Unable to start kubectl tunnel: {exc}', err=True, fg='red')
+                try:
+                    if self.kubectl_tunnel:
+                        self.kubectl_tunnel.close()
+                except Exception:
+                    pass
+                sys.exit(1)
         elif boundary_target_id:
             use_keyring = False
             boundary_executable = self.config.get('boundary', {}).get('boundary_executable', 'boundary') or 'boundary'
@@ -267,6 +294,21 @@ def connect(
                 vault_password_field=vault_password_field,
                 vault_username_field=vault_username_field,
             )
+        elif self.kubectl_tunnel:
+            display_dsn = format_connection_dsn(
+                user=display_dsn_user,
+                host=host,
+                port=int_port,
+                socket=None,
+                database=database,
+                character_set=character_set,
+                kubectl_resource=kubectl_resource,
+                vault_address=vault_address,
+                vault_mount=vault_mount,
+                vault_secret=vault_secret,
+                vault_password_field=vault_password_field,
+                vault_username_field=vault_username_field,
+            )
         elif self.boundary_tunnel:
             display_dsn = format_connection_dsn(
                 user=None,
@@ -312,6 +354,10 @@ def connect(
             connection_info['host'] = self.ssh_tunnel.local_host
             connection_info['port'] = self.ssh_tunnel.local_port
             connection_info['socket'] = None
+        elif self.kubectl_tunnel:
+            connection_info['host'] = self.kubectl_tunnel.local_host
+            connection_info['port'] = self.kubectl_tunnel.local_port
+            connection_info['socket'] = None
         elif self.boundary_tunnel:
             connection_info['user'] = self.boundary_tunnel.username
             connection_info['host'] = self.boundary_tunnel.local_host

mycli/constants.py

@@ -8,6 +8,7 @@
     'boundary_id',
     'character_set',
     'keepalive_ticks',
+    'kubectl_resource',
     'prompt',
     'socket',
     'ssh_jump',

mycli/kubectl_tunnel.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import re
+import shlex
+import subprocess
+import tempfile
+import time
+from typing import IO
+
+from mycli.compat import WIN
+
+DEFAULT_KUBECTL_EXECUTABLE = 'kubectl'
+LOCAL_HOST = '127.0.0.1'
+
+
+class KubectlTunnelError(RuntimeError):
+    pass
+
+
+def _split_options(options: str | None) -> list[str]:
+    try:
+        arguments = shlex.split(options or '', posix=not WIN)
+    except ValueError as exc:
+        raise KubectlTunnelError(f'Unable to parse kubectl options: {exc}') from exc
+    if WIN:
+        arguments = [
+            argument[1:-1] if len(argument) >= 2 and argument[0] == argument[-1] and argument[0] in ('"', "'") else argument
+            for argument in arguments
+        ]
+    return arguments
+
+
+class KubectlTunnel:
+    def __init__(
+        self,
+        *,
+        resource: str,
+        remote_port: int,
+        kubectl_executable: str = DEFAULT_KUBECTL_EXECUTABLE,
+        kubectl_config_options: str | None = None,
+        kubectl_cli_options: str | None = None,
+        local_port: int | None = None,
+        ready_timeout: float = 30.0,
+    ) -> None:
+        if not resource.strip():
+            raise KubectlTunnelError('Kubernetes resource must not be empty.')
+        if not 1 <= remote_port <= 65535:
+            raise KubectlTunnelError('Kubernetes remote port must be an integer between 1 and 65535.')
+        self.resource = resource
+        self.remote_port = remote_port
+        self.kubectl_executable = kubectl_executable
+        self.kubectl_config_options = kubectl_config_options
+        self.kubectl_cli_options = kubectl_cli_options
+        self.local_host = LOCAL_HOST
+        self.local_port = local_port
+        self.ready_timeout = ready_timeout
+        self.process: subprocess.Popen | None = None
+        self._output_file: IO[bytes] | None = None
+
+    def command(self) -> list[str]:
+        options = _split_options(self.kubectl_config_options)
+        options.extend(_split_options(self.kubectl_cli_options))
+        return [
+            self.kubectl_executable,
+            'port-forward',
+            *options,
+            f'--address={self.local_host}',
+            self.resource,
+            f'{self.local_port or ""}:{self.remote_port}',
+        ]
+
+    def start(self) -> None:
+        self._output_file = tempfile.TemporaryFile(mode='w+b')
+        try:
+            self.process = subprocess.Popen(
+                self.command(),
+                stdin=subprocess.DEVNULL,
+                stdout=self._output_file,
+                stderr=self._output_file,
+                start_new_session=True,
+            )
+        except KubectlTunnelError:
+            self.close()
+            raise
+        except (OSError, ValueError) as exc:
+            self.close()
+            raise KubectlTunnelError(f'Unable to start kubectl port-forward process: {exc}') from exc
+
+        deadline = time.monotonic() + self.ready_timeout
+        while time.monotonic() < deadline:
+            return_code = self.process.poll()
+            if return_code is not None:
+                output = self._captured_output()
+                self.close()
+                detail = f': {output}' if output else ''
+                raise KubectlTunnelError(f'kubectl port-forward exited with status {return_code}{detail}')
+            if local_port := self._forwarded_local_port():
+                self.local_port = local_port
+                return
+            time.sleep(0.05)
+
+        output = self._captured_output()
+        self.close()
+        detail = f': {output}' if output else ''
+        raise KubectlTunnelError(f'Timed out waiting for kubectl port-forward to become ready{detail}')
+
+    def close(self) -> None:
+        process = self.process
+        if process is not None and process.poll() is None:
+            process.terminate()
+            try:
+                process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+        if self._output_file is not None:
+            self._output_file.close()
+            self._output_file = None
+
+    def _captured_output(self) -> str:
+        if self._output_file is None:
+            return ''
+        self._output_file.seek(0)
+        return self._output_file.read().decode('utf-8', errors='replace').strip()
+
+    def _forwarded_local_port(self) -> int | None:
+        match = re.search(rf'Forwarding from {re.escape(self.local_host)}:(\d+) ->', self._captured_output())
+        return int(match.group(1)) if match else None

mycli/main.py

@@ -321,6 +321,14 @@ class CliArgs:
         type=str,
         help='Extra CLI arguments for SSH with --ssh-jump, placed after options from myclirc.',
     )
+    kubectl_resource: str | None = clickdc.option(
+        type=str,
+        help='Open a kubectl port-forward tunnel to RESOURCE and connect through it.',
+    )
+    kubectl_options: str | None = clickdc.option(
+        type=str,
+        help='Extra CLI arguments for kubectl with --kubectl-resource.',
+    )
     boundary_id: str | None = clickdc.option(
         type=str,
         help='Open a HashiCorp Boundary tunnel to TARGET_ID and connect through it.',

mycli/myclirc

@@ -414,6 +414,14 @@ ssh_options = -a -o ServerAliveInterval=60 -o ExitOnForwardFailure=yes -o IPQoS=
 # auto means: port if on Windows, socket otherwise.
 tunnel_method = auto
 
+[kubectl]
+# Path to the kubectl executable used by --kubectl-resource.
+kubectl_executable = kubectl
+
+# Options to pass to "kubectl port-forward" before the resource name.
+# This can be used, for example, to set the context.
+kubectl_options =
+
 [vault]
 # Path to the vault executable used for Vault integration.
 vault_executable = vault

mycli/packages/special/utils.py

@@ -156,6 +156,7 @@ def format_connection_dsn(
     socket: str | None,
     character_set: str | None,
     boundary_id: str | None = None,
+    kubectl_resource: str | None = None,
     ssh_jump: str | None = None,
     vault_address: str | None = None,
     vault_mount: str | None = None,
@@ -177,6 +178,8 @@ def format_connection_dsn(
         query_part['character_set'] = character_set
     if boundary_id:
         query_part['boundary_id'] = boundary_id
+    if kubectl_resource:
+        query_part['kubectl_resource'] = kubectl_resource
     if ssh_jump:
         query_part['ssh_jump'] = ssh_jump
     if vault_address:

mycli/resources/completions/bash/mycli

@@ -37,7 +37,7 @@ _mycli_is_positional_db_arg() {
                 ;;
             --*=*)
                 ;;
-            --host|--hostname|--port|--user|--username|--socket|--pass|--password|--password-file|--vault-address|--vault-mount|--vault-secret|--vault-password-field|--vault-username-field|--ssl-mode|--ssl-ca|--ssl-capath|--ssl-cert|--ssl-key|--ssl-cipher|--tls-version|--database|--dsn|--completions|--prompt|--toolbar|--logfile|--checkpoint|--myclirc|--local-infile|--login-path|--execute|--init-command|--charset|--character-set|--batch|--format|--throttle|--use-keyring|--keepalive-ticks|--ssh-jump|--ssh-options|--boundary-id)
+            --host|--hostname|--port|--user|--username|--socket|--pass|--password|--password-file|--vault-address|--vault-mount|--vault-secret|--vault-password-field|--vault-username-field|--ssl-mode|--ssl-ca|--ssl-capath|--ssl-cert|--ssl-key|--ssl-cipher|--tls-version|--database|--dsn|--completions|--prompt|--toolbar|--logfile|--checkpoint|--myclirc|--local-infile|--login-path|--execute|--init-command|--charset|--character-set|--batch|--format|--throttle|--use-keyring|--keepalive-ticks|--ssh-jump|--ssh-options|--kubectl-resource|--kubectl-options|--boundary-id)
                 expects_value=1
                 ;;
             -?*)

mycli/resources/completions/fish/mycli.fish

@@ -28,7 +28,7 @@ function _mycli_is_positional_db_arg
             case --
                 set options_ended 1
             case '--*=*'
-            case --host --hostname --port --user --username --socket --pass --password --password-file --vault-address --vault-mount --vault-secret --vault-password-field --vault-username-field --ssl-mode --ssl-ca --ssl-capath --ssl-cert --ssl-key --ssl-cipher --tls-version --database --dsn --completions --prompt --toolbar --logfile --checkpoint --myclirc --local-infile --login-path --execute --init-command --charset --character-set --batch --format --throttle --use-keyring --keepalive-ticks --ssh-jump --ssh-options --boundary-id
+            case --host --hostname --port --user --username --socket --pass --password --password-file --vault-address --vault-mount --vault-secret --vault-password-field --vault-username-field --ssl-mode --ssl-ca --ssl-capath --ssl-cert --ssl-key --ssl-cipher --tls-version --database --dsn --completions --prompt --toolbar --logfile --checkpoint --myclirc --local-infile --login-path --execute --init-command --charset --character-set --batch --format --throttle --use-keyring --keepalive-ticks --ssh-jump --ssh-options --kubectl-resource --kubectl-options --boundary-id
                 set expects_value 1
             case '-*'
                 set -l option_length (string length -- "$word")

mycli/resources/completions/zsh/_mycli

@@ -35,7 +35,7 @@ _mycli_is_positional_db_arg() {
                 ;;
             --*=*)
                 ;;
-            --host|--hostname|--port|--user|--username|--socket|--pass|--password|--password-file|--vault-address|--vault-mount|--vault-secret|--vault-password-field|--vault-username-field|--ssl-mode|--ssl-ca|--ssl-capath|--ssl-cert|--ssl-key|--ssl-cipher|--tls-version|--database|--dsn|--completions|--prompt|--toolbar|--logfile|--checkpoint|--myclirc|--local-infile|--login-path|--execute|--init-command|--charset|--character-set|--batch|--format|--throttle|--use-keyring|--keepalive-ticks|--ssh-jump|--ssh-options|--boundary-id)
+            --host|--hostname|--port|--user|--username|--socket|--pass|--password|--password-file|--vault-address|--vault-mount|--vault-secret|--vault-password-field|--vault-username-field|--ssl-mode|--ssl-ca|--ssl-capath|--ssl-cert|--ssl-key|--ssl-cipher|--tls-version|--database|--dsn|--completions|--prompt|--toolbar|--logfile|--checkpoint|--myclirc|--local-infile|--login-path|--execute|--init-command|--charset|--character-set|--batch|--format|--throttle|--use-keyring|--keepalive-ticks|--ssh-jump|--ssh-options|--kubectl-resource|--kubectl-options|--boundary-id)
                 expects_value=1
                 ;;
             -?*)