Add support for connecting via a Boundary tunnel

AGENTS.md

@@ -11,6 +11,7 @@ A command line client for MySQL with auto-completion and syntax highlighting.
 ├── mycli/                                # application source
 ├── mycli/__init__.py                     # provides version number
 ├── mycli/app_state.py                    # `AppStateMixin` application state mixin and related functions
+├── mycli/boundary_tunnel.py              # connection over Boundary tunnel
 ├── mycli/cli_runner.py                   # connects and dispatches main modes based on CLI arguments
 ├── mycli/clibuffer.py                    # prompt_toolkit buffer utilities
 ├── mycli/client_commands.py              # special commands which must be registered separately

changelog.md

@@ -1,3 +1,11 @@
+Upcoming (TBD)
+==============
+
+Features
+---------
+* Add beta support for HashiCorp Boundary tunnels.
+
+
 2.12.0 (2026/08/08)
 ==============
 

mycli/boundary_tunnel.py

@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+import datetime
+import json
+import os
+import shlex
+import socket
+import subprocess
+import threading
+import time
+
+
+class BoundaryTunnelError(RuntimeError):
+    pass
+
+
+def _find_free_local_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(('127.0.0.1', 0))
+        return int(sock.getsockname()[1])
+
+
+class BoundaryTunnel:
+    def __init__(
+        self,
+        *,
+        target_id: str,
+        boundary_executable: str = 'boundary',
+        address: str | None = None,
+        auth_method_id: str | None = None,
+        boundary_options: str | None = None,
+        local_port: int | None = None,
+        ready_timeout: float = 30.0,
+    ) -> None:
+        self.target_id = target_id
+        self.boundary_executable = boundary_executable
+        self.address = address
+        self.auth_method_id = auth_method_id
+        self.boundary_options = boundary_options
+        self.local_host = '127.0.0.1'
+        self.local_port = local_port or _find_free_local_port()
+        self.ready_timeout = ready_timeout
+        self.process: subprocess.Popen | None = None
+        self.stdout = ''
+        self._startup_error: OSError | ValueError | None = None
+        self._started = threading.Event()
+        self._ready = threading.Event()
+        self._failed = threading.Event()
+        self._thread: threading.Thread | None = None
+        self.username: str | None = None
+        self.password: str | None = None
+        self.expiry: str | None = None
+
+    def command(self) -> list[str]:
+        options = shlex.split(self.boundary_options or '')
+        command = [
+            self.boundary_executable,
+            'connect',
+            *options,
+            f'-target-id={self.target_id}',
+            f'-listen-addr={self.local_host}',
+            f'-listen-port={self.local_port}',
+            '-format=json',
+        ]
+        if self.address:
+            command.append(f'-addr={self.address}')
+        return command
+
+    def start(self, *, show_expiration_warning: bool = True) -> None:
+        self._thread = threading.Thread(target=self._run, name='mycli-boundary-tunnel', daemon=True)
+        self._thread.start()
+        deadline = time.monotonic() + self.ready_timeout
+        while time.monotonic() < deadline:
+            if self._failed.is_set():
+                self.close()
+                if self._startup_error is not None:
+                    raise BoundaryTunnelError(f'Unable to start Boundary tunnel process: {self._startup_error}') from self._startup_error
+                raise BoundaryTunnelError('Boundary tunnel process exited before it was ready.')
+            if self._started.is_set():
+                self._read_stdout()
+                break
+            time.sleep(0.05)
+        else:
+            self.close()
+            raise BoundaryTunnelError('Timed out waiting for Boundary tunnel process to start.')
+
+        connection_details = json.loads(self.stdout)
+        self.username = connection_details['credentials'][0]['secret']['decoded']['username']
+        self.password = connection_details['credentials'][0]['secret']['decoded']['password']
+        expiry_raw = connection_details['expiration']
+        expiry_utc = datetime.datetime.strptime(expiry_raw, '%Y-%m-%dT%H:%M:%S.%f%z')
+        expiry_local = datetime.datetime.fromtimestamp(expiry_utc.timestamp())
+        self.expiry = datetime.datetime.strftime(expiry_local, '%H:%M:%S %a %d %b %Y')
+
+        while time.monotonic() < deadline:
+            if self._failed.is_set():
+                self.close()
+                if self._startup_error is not None:
+                    raise BoundaryTunnelError(f'Unable to start Boundary tunnel process: {self._startup_error}') from self._startup_error
+                raise BoundaryTunnelError('Boundary tunnel process exited before it was ready.')
+            if self._is_listening():
+                self._ready.set()
+                return
+            time.sleep(0.05)
+        self.close()
+        raise BoundaryTunnelError('Timed out waiting for Boundary tunnel to become ready.')
+
+    def _read_stdout(self) -> None:
+        process = self.process
+        if process is None or process.stdout is None:
+            return
+        self.stdout = process.stdout.readline().decode('utf-8')
+
+    def _environment(self) -> dict[str, str] | None:
+        if not self.auth_method_id:
+            return None
+        environment = os.environ.copy()
+        environment['BOUNDARY_AUTH_METHOD_ID'] = self.auth_method_id
+        return environment
+
+    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._thread is not None and self._thread.is_alive():
+            self._thread.join(timeout=5)
+
+    def _run(self) -> None:
+        try:
+            self.process = subprocess.Popen(
+                self.command(),
+                stdin=subprocess.DEVNULL,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.DEVNULL,
+                start_new_session=True,
+                env=self._environment(),
+            )
+            self._started.set()
+        except (OSError, ValueError) as exc:
+            self._startup_error = exc
+            self._failed.set()
+            return
+        return_code = self.process.wait()
+        if return_code != 0 and not self._ready.is_set():
+            self._failed.set()
+
+    def _is_listening(self) -> bool:
+        try:
+            with socket.create_connection((self.local_host, self.local_port), timeout=0.05):
+                return True
+        except OSError:
+            return False

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('boundary_id'):
+            cli_args.boundary_id = cli_args.boundary_id or params[0]
         if params := dsn_params.get('vault_address'):
             cli_args.vault_address = cli_args.vault_address or params[0]
         if params := dsn_params.get('vault_mount'):
@@ -521,6 +523,7 @@ def load_vault_password() -> str | None:
             vault_password_field=cli_args.vault_password_field,
             vault_username_field=cli_args.vault_username_field,
             vault_username_from_vault=vault_username_from_vault,
+            boundary_target_id=cli_args.boundary_id,
         )
 
         if combined_init_cmd:

mycli/client.py

@@ -20,6 +20,7 @@
     normalize_image_protocol,
     normalize_ssl_mode,
 )
+from mycli.boundary_tunnel import BoundaryTunnel
 from mycli.client_commands import ClientCommandsMixin, get_config_property_names
 from mycli.client_connection import ClientConnectionMixin
 from mycli.client_query import ClientQueryMixin
@@ -79,6 +80,7 @@ def __init__(
     ) -> None:
         self.sqlexecute = sqlexecute
         self.ssh_tunnel: SshTunnel | None = None
+        self.boundary_tunnel: BoundaryTunnel | None = None
         self.logfile = logfile
         self.login_path = login_path
         self.toolbar_error_message: str | None = None
@@ -253,6 +255,11 @@ def close(self) -> None:
                 self.ssh_tunnel.close()
             except Exception:
                 pass
+        if self.boundary_tunnel:
+            try:
+                self.boundary_tunnel.close()
+            except Exception:
+                pass
 
     def run_cli(self) -> None:
         repl_package.main_repl(self)

mycli/client_connection.py

@@ -12,6 +12,7 @@
 from pymysql.constants.CR import CR_SERVER_LOST
 from pymysql.constants.ER import ACCESS_DENIED_ERROR, HANDSHAKE_ERROR
 
+from mycli.boundary_tunnel import BoundaryTunnel, BoundaryTunnelError
 from mycli.compat import WIN
 from mycli.config import str_to_bool
 from mycli.constants import (
@@ -40,8 +41,10 @@ class ClientConnectionMixin:
         config_without_package_defaults: Any
         keepalive_ticks: int | None
         sandbox_mode: bool
+        verbosity: int
         sqlexecute: Any
         logger: Any
+        boundary_tunnel: BoundaryTunnel | None
 
         def read_mylogin_cnf(self, cnf: Any) -> dict[str, Any]: ...
         def echo(self, *args: Any, **kwargs: Any) -> None: ...
@@ -70,6 +73,7 @@ def connect(
         vault_password_field: str | None = None,
         vault_username_field: str | None = None,
         vault_username_from_vault: bool = False,
+        boundary_target_id: str | None = None,
     ) -> None:
         mylogin_cnf: dict[str, Any] = self.read_mylogin_cnf(self.mylogin_cnf)
         # Fall back to .mylogin.cnf values only if user did not specify a value.
@@ -81,6 +85,7 @@ def connect(
         self.keepalive_ticks = keepalive_ticks
         self.ssh_tunnel = None
         self.selected_password = None
+        self.boundary_tunnel = None
 
         int_port = port and int(port)
         if not int_port:
@@ -125,9 +130,48 @@ def connect(
                 except Exception:
                     pass
                 sys.exit(1)
+        elif boundary_target_id:
+            use_keyring = False
+            boundary_executable = self.config.get('boundary_beta', {}).get('boundary_executable', 'boundary') or 'boundary'
+            boundary_address = self.config.get('boundary_beta', {}).get('address') or None
+            boundary_auth_method_id = self.config.get('boundary_beta', {}).get('auth_method_id') or None
+            boundary_options = self.config.get('boundary_beta', {}).get('boundary_options') or None
+            try:
+                self.boundary_tunnel = BoundaryTunnel(
+                    target_id=boundary_target_id,
+                    boundary_executable=boundary_executable,
+                    address=boundary_address,
+                    auth_method_id=boundary_auth_method_id,
+                    boundary_options=boundary_options,
+                )
+                self.boundary_tunnel.start(show_expiration_warning=self.verbosity >= 0)
+                if self.verbosity >= 0:
+                    click.secho(
+                        f'The Boundary db connection will expire at: {self.boundary_tunnel.expiry}',
+                        fg='white',
+                        bg='red',
+                        bold=True,
+                        err=True,
+                    )
+                if self.verbosity >= 2:
+                    click.secho(f'Temporary username: {self.boundary_tunnel.username}', err=True)
+                    click.secho(f'Temporary password: {self.boundary_tunnel.password}', err=True)
+                    click.secho(f'Temporary host:     {self.boundary_tunnel.local_host}', err=True)
+                    click.secho(f'Temporary port:     {self.boundary_tunnel.local_port}', err=True)
+                socket = None
+            except (OSError, BoundaryTunnelError) as exc:
+                click.secho(f'Error: Unable to start Boundary tunnel: {exc}', err=True, fg='red')
+                try:
+                    if self.boundary_tunnel:
+                        self.boundary_tunnel.close()
+                except Exception:
+                    pass
+                sys.exit(1)
 
         if password_candidates is None:
             password_candidates = PasswordCandidates()
+        if self.boundary_tunnel:
+            password_candidates.add_value('boundary', self.boundary_tunnel.password)
 
         if not character_set:
             if 'main' in self.config_without_package_defaults and 'default_character_set' in self.config_without_package_defaults['main']:
@@ -219,6 +263,16 @@ def connect(
                 vault_password_field=vault_password_field,
                 vault_username_field=vault_username_field,
             )
+        elif self.boundary_tunnel:
+            display_dsn = format_connection_dsn(
+                user=None,
+                host=host,
+                port=None,
+                socket=None,
+                database=database,
+                character_set=character_set,
+                boundary_id=boundary_target_id,
+            )
         elif vault_secret:
             display_dsn = format_connection_dsn(
                 user=display_dsn_user,
@@ -254,6 +308,11 @@ def connect(
             connection_info['host'] = self.ssh_tunnel.local_host
             connection_info['port'] = self.ssh_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
+            connection_info['port'] = self.boundary_tunnel.local_port
+            connection_info['socket'] = None
         else:
             connection_info['host'] = host
             connection_info['port'] = int_port

mycli/constants.py

@@ -5,6 +5,7 @@
 
 DEFAULT_CHARSET = 'utf8mb4'
 KNOWN_DSN_QUERY_PARAMS = {
+    'boundary_id',
     'character_set',
     'keepalive_ticks',
     'prompt',

mycli/main.py

@@ -321,6 +321,11 @@ class CliArgs:
         type=str,
         help='Extra CLI arguments for SSH with --ssh-jump, placed after options from myclirc.',
     )
+    boundary_id: str | None = clickdc.option(
+        type=str,
+        help='BETA: open a HashiCorp Boundary tunnel to TARGET_ID and connect through it.',
+        hidden=True,
+    )
     checkup: bool = clickdc.option(
         is_flag=True,
         help='Run a checkup on your configuration.',

mycli/myclirc

@@ -233,7 +233,7 @@ pager = 'less'
 # * file           # --password-file=<file> at the CLI
 # * environment    # $MYSQL_PWD environment variable
 # * dsn            # mysql://user:password@ field in a DSN
-# * boundary       # value provided by Boundary client (not yet implemented)
+# * boundary       # value provided by Boundary client
 # * vault          # value retrieved from "vault kv get"
 # * login_path     # --login-path=<path> at the CLI
 # * keyring        # value retrieved from system keyring
@@ -422,6 +422,19 @@ default_password_field = password
 # Field/property containing the username, if --vault-username-field is not provided.
 default_username_field = username
 
+[boundary_beta]
+# Path to the HashiCorp Boundary executable used by --boundary-id.
+boundary_executable = boundary
+
+# Address of the Boundary controller. If empty, use Boundary's default resolution.
+address =
+
+# Boundary auth method ID. If empty, use BOUNDARY_AUTH_METHOD_ID from the environment.
+auth_method_id =
+
+# Additional options passed to the "boundary connect" command.
+boundary_options =
+
 # Custom colors for the completion menu, toolbar, etc, with actual support
 # depending on the terminal, and the property being set.
 # Colors: #ffffff, bg:#ffffff, border:#ffffff.