Ability to read passwords from Vault service

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
 

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

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,

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:

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.',

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.

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')

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.

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'

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'),
     [