From f6cdf07248eab8cc6f7fce947191ec2206fc0268 Mon Sep 17 00:00:00 2001 From: Roland Walker Date: Sat, 11 Jul 2026 10:30:17 -0400 Subject: [PATCH] ability to read passwords from Vault service via the CLI interface "vault kv get". The user must have already authenticated with "vault login". Several CLI arguments are added: * --password-vault-address * --password-vault-mount * --password-vault-field * --password-vault-secret but only --password-vault-secret is mandatory, though --password-vault-field may also be often needed. This is documented as experimental functionality. --- AGENTS.md | 5 +- changelog.md | 1 + mycli/cli_runner.py | 23 +++++- mycli/client_connection.py | 5 +- mycli/main.py | 16 ++++ mycli/myclirc | 13 +++ mycli/vault.py | 54 ++++++++++++ test/myclirc | 13 +++ test/pytests/test_cli_runner.py | 142 ++++++++++++++++++++++++++++++++ test/pytests/test_main.py | 25 ++++++ test/pytests/test_vault.py | 96 +++++++++++++++++++++ 11 files changed, 388 insertions(+), 5 deletions(-) create mode 100644 mycli/vault.py create mode 100644 test/pytests/test_vault.py diff --git a/AGENTS.md b/AGENTS.md index 52ebdbea..5632c4ec 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -48,13 +48,14 @@ A command line client for MySQL with auto-completion and syntax highlighting. ├── mycli/schema_prefetcher.py # background prefetcher for multi-schema auto-completion ├── mycli/sqlcompleter.py # offers SQL completions ├── mycli/sqlexecute.py # runs SQL queries +├── mycli/types.py # shared types +├── mycli/vault.py # Vault integration ├── test/conftest.py # pytest configuration ├── test/features/ # behave tests ├── test/myclirc # mycli configuration used for tests ├── test/mylogin.cnf # `mylogin.cnf` example used for tests ├── test/pytests/ # pytest tests -├── test/utils.py # shared utilities for tests -└── types.py # shared types +└── test/utils.py # shared utilities for tests ## Development diff --git a/changelog.md b/changelog.md index df6ee157..d988219f 100644 --- a/changelog.md +++ b/changelog.md @@ -11,6 +11,7 @@ Features * Add `--ssh-options` CLI argument to pass extra options with `--ssh-jump`. * Make `ssh_jump` a DSN query parameter. * Warn on unknown DSN query parameters. +* Experimental: ability to read passwords from Vault using `vault kv get` CLI. Bug Fixes diff --git a/mycli/cli_runner.py b/mycli/cli_runner.py index 313a0476..9d5136f0 100644 --- a/mycli/cli_runner.py +++ b/mycli/cli_runner.py @@ -15,6 +15,7 @@ from mycli.main_modes.execute import main_execute_from_cli from mycli.main_modes.list_dsn import main_list_dsn from mycli.packages.cli_utils import is_valid_connection_scheme +from mycli.vault import DEFAULT_VAULT_EXECUTABLE, DEFAULT_VAULT_FIELD, VaultError, get_password_from_vault if TYPE_CHECKING: from mycli.main import CliArgs @@ -342,11 +343,31 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non use_keyring = str_to_bool(cli_args.use_keyring) reset_keyring = False + if cli_args.password is None and cli_args.password_vault_secret: + vault_config = mycli.config.get('vault', {}) + vault_address = cli_args.password_vault_address or os.environ.get('VAULT_ADDR') or vault_config.get('address') or None + vault_mount = cli_args.password_vault_mount or vault_config.get('default_mount') or None + vault_field = cli_args.password_vault_field or vault_config.get('default_field') or DEFAULT_VAULT_FIELD + vault_executable = vault_config.get('vault_executable') or DEFAULT_VAULT_EXECUTABLE + try: + vault_password: str | None = get_password_from_vault( + secret=cli_args.password_vault_secret, + executable=vault_executable, + field=vault_field, + mount=vault_mount, + address=vault_address, + ) + except VaultError as exc: + click.secho(f'Error reading password from Vault: {exc}', err=True, fg='red') + sys.exit(1) + else: + vault_password = None + try: mycli.connect( database=database, user=cli_args.user, - passwd=cli_args.password, + passwd=vault_password if cli_args.password is None else cli_args.password, host=cli_args.host, port=cli_args.port, socket=cli_args.socket, diff --git a/mycli/client_connection.py b/mycli/client_connection.py index 300b5e15..04c4019f 100644 --- a/mycli/client_connection.py +++ b/mycli/client_connection.py @@ -166,8 +166,9 @@ def connect( # 2. --password-file CLI option # 3. envvar (MYSQL_PWD) # 4. DSN (mysql://user:password) - # 5. .mylogin.cnf - # 6. keyring + # 5. Vault + # 6. .mylogin.cnf + # 7. keyring ssh_tunnel_field = urlquote(ssh_jump or '') if ssh_tunnel_field: diff --git a/mycli/main.py b/mycli/main.py index f3cf66e5..be2231b7 100755 --- a/mycli/main.py +++ b/mycli/main.py @@ -89,6 +89,22 @@ class CliArgs: type=click.Path(), help='File or FIFO path containing the password to connect to the db if not specified otherwise.', ) + password_vault_address: str | None = clickdc.option( + type=str, + help='EXPERIMENTAL "vault kv get" integration: value for $VAULT_ADDR if unset in the environment or ~/.myclirc.', + ) + password_vault_mount: str | None = clickdc.option( + type=str, + help='EXPERIMENTAL "vault kv get" integration: value for -mount if unset in ~/.myclirc.', + ) + password_vault_field: str | None = clickdc.option( + type=str, + help='EXPERIMENTAL "vault kv get" integration: value for -field if unset in ~/.myclirc.', + ) + password_vault_secret: str | None = clickdc.option( + type=str, + help='EXPERIMENTAL "vault kv get" integration: secret name.', + ) ssl_mode: str = clickdc.option( type=click.Choice(['auto', 'on', 'off']), help='Set desired SSL behavior. auto=preferred if TCP/IP, on=required, off=off.', diff --git a/mycli/myclirc b/mycli/myclirc index 4bfb6e3a..0fb87ad6 100644 --- a/mycli/myclirc +++ b/mycli/myclirc @@ -317,6 +317,19 @@ ssh_options = -a -o ServerAliveInterval=60 -o ExitOnForwardFailure=yes -o IPQoS= # auto means: port if on Windows, socket otherwise. tunnel_method = auto +[vault] +# Path to the vault executable used for Vault integration. +vault_executable = vault + +# URL for VAULT_ADDR, if the environment variable is unset. +address = + +# Default mount, which can be overridden by --password-vault-mount at the CLI. +default_mount = + +# Field/property containing the password, if --password-vault-field is not provided. +default_field = password + # 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. diff --git a/mycli/vault.py b/mycli/vault.py new file mode 100644 index 00000000..6f39ae6e --- /dev/null +++ b/mycli/vault.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import subprocess + +DEFAULT_VAULT_EXECUTABLE = 'vault' +DEFAULT_VAULT_FIELD = 'password' + + +class VaultError(RuntimeError): + pass + + +def get_password_from_vault( + *, + secret: str, + executable: str = DEFAULT_VAULT_EXECUTABLE, + field: str = DEFAULT_VAULT_FIELD, + mount: str | None = None, + address: str | None = None, +) -> str: + command = [ + executable, + 'kv', + 'get', + f'-field={field}', + ] + if mount: + command.append(f'-mount={mount}') + if address: + command.append(f'-address={address}') + + command.append(secret) + + try: + completed_process = subprocess.run( + command, + check=False, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + except FileNotFoundError as exc: + raise VaultError(f'Vault executable not found: {executable}') from exc + except OSError as exc: + raise VaultError(f'Unable to run Vault executable {executable}: {exc}') from exc + + if completed_process.returncode: + stderr = completed_process.stderr.strip() + if stderr: + raise VaultError(f'Vault command failed. You may need to run "vault login": {stderr}') + raise VaultError(f'Vault command failed. You may need to run "vault login". Exit code {completed_process.returncode}.') + + return completed_process.stdout.removesuffix('\n') diff --git a/test/myclirc b/test/myclirc index 4ee66152..1ae41994 100644 --- a/test/myclirc +++ b/test/myclirc @@ -317,6 +317,19 @@ ssh_options = -a -o ServerAliveInterval=60 -o ExitOnForwardFailure=yes -o IPQoS= # auto means: port if on Windows, socket otherwise. tunnel_method = auto +[vault] +# Path to the vault executable used for Vault integration. +vault_executable = vault + +# URL for VAULT_ADDR, if the environment variable and CLI argument are unset. +address = + +# Default mount, which can be overridden by --password-vault-mount at the CLI. +default_mount = + +# Field/property containing the password, if --password-vault-field is not provided. +default_field = password + # 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. diff --git a/test/pytests/test_cli_runner.py b/test/pytests/test_cli_runner.py index 110aec70..9e15034e 100644 --- a/test/pytests/test_cli_runner.py +++ b/test/pytests/test_cli_runner.py @@ -52,6 +52,7 @@ def default_config() -> dict[str, Any]: return { 'main': {'use_keyring': 'false'}, 'connection': {'default_keepalive_ticks': 0}, + 'vault': {}, 'alias_dsn': {}, 'init-commands': {}, 'alias_dsn.init-commands': {}, @@ -540,6 +541,147 @@ def test_run_from_cli_args_uses_explicit_keyring_flag(monkeypatch: pytest.Monkey assert client.connect_calls[-1]['reset_keyring'] is False +def test_run_from_cli_args_reads_password_from_vault_when_password_is_missing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + cli_args = make_cli_args() + cli_args.password_vault_secret = 'database/prod' + client = DummyMyCli( + config={ + **default_config(), + 'vault': { + 'vault_executable': '/opt/bin/vault', + 'address': 'https://vault.config', + 'default_mount': 'kv', + 'default_field': 'mysql_password', + }, + } + ) + vault_calls: list[dict[str, str | None]] = [] + + def fake_get_password_from_vault(**kwargs: str | None) -> str: + vault_calls.append(kwargs) + return 'vault-secret' + + monkeypatch.delenv('VAULT_ADDR', raising=False) + monkeypatch.setattr(cli_runner, 'get_password_from_vault', fake_get_password_from_vault) + + run_with_client(monkeypatch, cli_args, client) + + assert client.connect_calls[-1]['passwd'] == 'vault-secret' + assert vault_calls == [ + { + 'secret': 'database/prod', + 'executable': '/opt/bin/vault', + 'field': 'mysql_password', + 'mount': 'kv', + 'address': 'https://vault.config', + } + ] + + +def test_run_from_cli_args_prefers_dsn_password_over_vault(monkeypatch: pytest.MonkeyPatch) -> None: + cli_args = make_cli_args() + cli_args.dsn = 'mysql://user:dsn-secret@host/db' + cli_args.password_vault_secret = 'database/prod' + client = DummyMyCli() + vault_calls: list[dict[str, str | None]] = [] + + def fake_get_password_from_vault(**kwargs: str | None) -> str: + vault_calls.append(kwargs) + return 'vault-secret' + + monkeypatch.setattr(cli_runner, 'get_password_from_vault', fake_get_password_from_vault) + + run_with_client(monkeypatch, cli_args, client) + + assert client.connect_calls[-1]['passwd'] == 'dsn-secret' + assert vault_calls == [] + + +def test_run_from_cli_args_prefers_vault_cli_values_and_env_address( + monkeypatch: pytest.MonkeyPatch, +) -> None: + cli_args = make_cli_args() + cli_args.password_vault_secret = 'database/prod' + cli_args.password_vault_mount = 'cli-mount' + cli_args.password_vault_field = 'cli-field' + client = DummyMyCli( + config={ + **default_config(), + 'vault': { + 'vault_executable': '/opt/bin/vault', + 'address': 'https://vault.config', + 'default_mount': 'config-mount', + 'default_field': 'config-field', + }, + } + ) + vault_calls: list[dict[str, str | None]] = [] + monkeypatch.setenv('VAULT_ADDR', 'https://vault.env') + + def fake_get_password_from_vault(**kwargs: str | None) -> str: + vault_calls.append(kwargs) + return 'vault-secret' + + monkeypatch.setattr(cli_runner, 'get_password_from_vault', fake_get_password_from_vault) + + run_with_client(monkeypatch, cli_args, client) + + assert client.connect_calls[-1]['passwd'] == 'vault-secret' + assert vault_calls == [ + { + 'secret': 'database/prod', + 'executable': '/opt/bin/vault', + 'field': 'cli-field', + 'mount': 'cli-mount', + 'address': 'https://vault.env', + } + ] + + +def test_run_from_cli_args_prefers_vault_cli_address_over_env( + monkeypatch: pytest.MonkeyPatch, +) -> None: + cli_args = make_cli_args() + cli_args.password_vault_secret = 'database/prod' + cli_args.password_vault_address = 'https://vault.cli' + client = DummyMyCli() + vault_calls: list[dict[str, str | None]] = [] + monkeypatch.setenv('VAULT_ADDR', 'https://vault.env') + + def fake_get_password_from_vault(**kwargs: str | None) -> str: + vault_calls.append(kwargs) + return 'vault-secret' + + monkeypatch.setattr(cli_runner, 'get_password_from_vault', fake_get_password_from_vault) + + run_with_client(monkeypatch, cli_args, client) + + assert client.connect_calls[-1]['passwd'] == 'vault-secret' + assert vault_calls[-1]['address'] == 'https://vault.cli' + + +def test_run_from_cli_args_reports_vault_error(monkeypatch: pytest.MonkeyPatch) -> None: + cli_args = make_cli_args() + cli_args.password_vault_secret = 'database/prod' + client = DummyMyCli() + secho_calls: list[tuple[str, dict[str, Any]]] = [] + monkeypatch.setattr(cli_runner.click, 'secho', lambda text, **kwargs: secho_calls.append((text, kwargs))) + monkeypatch.setattr( + cli_runner, + 'get_password_from_vault', + lambda **_kwargs: (_ for _ in ()).throw(cli_runner.VaultError('permission denied')), + ) + + with pytest.raises(SystemExit) as excinfo: + run_with_client(monkeypatch, cli_args, client) + + assert excinfo.value.code == 1 + assert client.connect_calls == [] + assert secho_calls == [('Error reading password from Vault: permission denied', {'err': True, 'fg': 'red'})] + + def test_run_from_cli_args_passes_ssh_options_to_connect(monkeypatch: pytest.MonkeyPatch) -> None: cli_args = make_cli_args() cli_args.ssh_options = '-o Compression=yes' diff --git a/test/pytests/test_main.py b/test/pytests/test_main.py index 1d2aa1fc..e46c22d1 100644 --- a/test/pytests/test_main.py +++ b/test/pytests/test_main.py @@ -2385,6 +2385,31 @@ def test_preprocess_cli_args_prefers_existing_password_over_mysql_pwd(monkeypatc assert cli_args.password == 'cli-secret' +def test_click_entrypoint_populates_password_vault_options(monkeypatch: pytest.MonkeyPatch) -> None: + cli_args_calls: list[CliArgs] = [] + monkeypatch.setattr(main, 'run_from_cli_args', lambda cli_args, client_factory: cli_args_calls.append(cli_args)) + + result = CliRunner().invoke( + click_entrypoint, + [ + '--password-vault-address', + 'https://vault.example.com', + '--password-vault-mount', + 'kv', + '--password-vault-secret', + 'database/prod', + '--password-vault-field', + 'mysql_password', + ], + ) + + assert result.exit_code == 0 + assert cli_args_calls[-1].password_vault_address == 'https://vault.example.com' + assert cli_args_calls[-1].password_vault_mount == 'kv' + assert cli_args_calls[-1].password_vault_secret == 'database/prod' + assert cli_args_calls[-1].password_vault_field == 'mysql_password' + + @pytest.mark.parametrize( ('checkpoint', 'batch', 'expected'), [ diff --git a/test/pytests/test_vault.py b/test/pytests/test_vault.py new file mode 100644 index 00000000..f3c3f398 --- /dev/null +++ b/test/pytests/test_vault.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +import subprocess +from types import SimpleNamespace +from typing import Any + +import pytest + +from mycli import vault + + +def test_get_password_from_vault_runs_kv_get_with_field_mount_and_address( + monkeypatch: pytest.MonkeyPatch, +) -> None: + run_calls: list[dict[str, Any]] = [] + + def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace: + run_calls.append({'command': command, **kwargs}) + return SimpleNamespace(returncode=0, stdout='secret\n', stderr='') + + monkeypatch.setattr(vault.subprocess, 'run', fake_run) + + password = vault.get_password_from_vault( + secret='database/prod', + executable='/opt/bin/vault', + field='mysql_password', + mount='kv', + address='https://vault.example.com', + ) + + assert password == 'secret' + assert run_calls == [ + { + 'command': [ + '/opt/bin/vault', + 'kv', + 'get', + '-field=mysql_password', + '-mount=kv', + '-address=https://vault.example.com', + 'database/prod', + ], + 'check': False, + 'stdin': subprocess.DEVNULL, + 'stdout': subprocess.PIPE, + 'stderr': subprocess.PIPE, + 'text': True, + } + ] + + +def test_get_password_from_vault_reports_missing_executable(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + raise FileNotFoundError() + + monkeypatch.setattr(vault.subprocess, 'run', fake_run) + + with pytest.raises(vault.VaultError, match='Vault executable not found: missing-vault'): + vault.get_password_from_vault(secret='database/prod', executable='missing-vault') + + +def test_get_password_from_vault_reports_oserror(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + raise OSError('boom') + + monkeypatch.setattr(vault.subprocess, 'run', fake_run) + + with pytest.raises(vault.VaultError, match='Unable to run Vault executable vault: boom'): + vault.get_password_from_vault(secret='database/prod') + + +def test_get_password_from_vault_reports_nonzero_exit_without_stdout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + return SimpleNamespace(returncode=2, stdout='secret\n', stderr='permission denied\n') + + monkeypatch.setattr(vault.subprocess, 'run', fake_run) + + with pytest.raises(vault.VaultError) as excinfo: + vault.get_password_from_vault(secret='database/prod') + + assert 'permission denied' in str(excinfo.value) + assert 'secret' not in str(excinfo.value) + + +def test_get_password_from_vault_reports_nonzero_exit_without_stderr( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + return SimpleNamespace(returncode=2, stdout='', stderr='') + + monkeypatch.setattr(vault.subprocess, 'run', fake_run) + + with pytest.raises(vault.VaultError, match='Vault command failed\\. You may need to run "vault login"\\. Exit code 2\\.'): + vault.get_password_from_vault(secret='database/prod')