diff --git a/changelog.md b/changelog.md index ea368483..752406d6 100644 --- a/changelog.md +++ b/changelog.md @@ -11,7 +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. +* Experimental: reading credentials from Vault using the `vault kv get` CLI. Bug Fixes diff --git a/mycli/cli_runner.py b/mycli/cli_runner.py index 0e02e102..20bbba5e 100644 --- a/mycli/cli_runner.py +++ b/mycli/cli_runner.py @@ -18,8 +18,9 @@ from mycli.vault import ( DEFAULT_VAULT_EXECUTABLE, DEFAULT_VAULT_PASSWORD_FIELD, + DEFAULT_VAULT_USERNAME_FIELD, VaultError, - get_password_from_vault, + get_field_from_vault, ) if TYPE_CHECKING: @@ -348,17 +349,17 @@ 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: + if cli_args.password is None and cli_args.vault_secret: vault_config = mycli.config.get('vault_beta', {}) - 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_password_field') or DEFAULT_VAULT_PASSWORD_FIELD + vault_address = cli_args.vault_address or os.environ.get('VAULT_ADDR') or vault_config.get('address') or None + vault_mount = cli_args.vault_mount or vault_config.get('default_mount') or None + vault_field = cli_args.vault_password_field or vault_config.get('default_password_field') or DEFAULT_VAULT_PASSWORD_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, + vault_password: str | None = get_field_from_vault( + vault_field, + cli_args.vault_secret, executable=vault_executable, - field=vault_field, mount=vault_mount, address=vault_address, ) @@ -368,10 +369,30 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non else: vault_password = None + if cli_args.user is None and cli_args.vault_secret: + vault_config = mycli.config.get('vault_beta', {}) + vault_address = cli_args.vault_address or os.environ.get('VAULT_ADDR') or vault_config.get('address') or None + vault_mount = cli_args.vault_mount or vault_config.get('default_mount') or None + vault_field = cli_args.vault_username_field or vault_config.get('default_username_field') or DEFAULT_VAULT_USERNAME_FIELD + vault_executable = vault_config.get('vault_executable') or DEFAULT_VAULT_EXECUTABLE + try: + vault_username = get_field_from_vault( + vault_field, + cli_args.vault_secret, + executable=vault_executable, + mount=vault_mount, + address=vault_address, + ) + except VaultError as exc: + click.secho(f'Error reading username from Vault: {exc}', err=True, fg='red') + sys.exit(1) + else: + vault_username = None + try: mycli.connect( database=database, - user=cli_args.user, + user=vault_username if cli_args.user is None else cli_args.user, passwd=vault_password if cli_args.password is None else cli_args.password, host=cli_args.host, port=cli_args.port, diff --git a/mycli/main.py b/mycli/main.py index be2231b7..d4a6811c 100755 --- a/mycli/main.py +++ b/mycli/main.py @@ -89,21 +89,25 @@ 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( + 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( + 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( + vault_secret: str | None = clickdc.option( type=str, - help='EXPERIMENTAL "vault kv get" integration: value for -field if unset in ~/.myclirc.', + help='EXPERIMENTAL "vault kv get" integration: secret name.', ) - password_vault_secret: str | None = clickdc.option( + vault_password_field: str | None = clickdc.option( type=str, - help='EXPERIMENTAL "vault kv get" integration: secret name.', + help='EXPERIMENTAL "vault kv get" integration: field containing the password.', + ) + vault_username_field: str | None = clickdc.option( + type=str, + help='EXPERIMENTAL "vault kv get" integration: field containing the username.', ) ssl_mode: str = clickdc.option( type=click.Choice(['auto', 'on', 'off']), diff --git a/mycli/myclirc b/mycli/myclirc index f1ce97ce..d684fcff 100644 --- a/mycli/myclirc +++ b/mycli/myclirc @@ -324,12 +324,15 @@ 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, which can be overridden by --vault-mount at the CLI. default_mount = -# Field/property containing the password, if --password-vault-field is not provided. +# Field/property containing the password, if --vault-password-field is not provided. default_password_field = password +# Field/property containing the username, if --vault-username-field is not provided. +default_username_field = username + # 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 index db89d354..c7ac2fdd 100644 --- a/mycli/vault.py +++ b/mycli/vault.py @@ -4,17 +4,17 @@ DEFAULT_VAULT_EXECUTABLE = 'vault' DEFAULT_VAULT_PASSWORD_FIELD = 'password' +DEFAULT_VAULT_USERNAME_FIELD = 'username' class VaultError(RuntimeError): pass -def get_password_from_vault( - *, +def get_field_from_vault( + field: str, secret: str, executable: str = DEFAULT_VAULT_EXECUTABLE, - field: str = DEFAULT_VAULT_PASSWORD_FIELD, mount: str | None = None, address: str | None = None, ) -> str: diff --git a/test/myclirc b/test/myclirc index d114cabf..a5c6b0a9 100644 --- a/test/myclirc +++ b/test/myclirc @@ -321,15 +321,18 @@ tunnel_method = auto # 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. +# 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, which can be overridden by --vault-mount at the CLI. default_mount = -# Field/property containing the password, if --password-vault-field is not provided. +# Field/property containing the password, if --vault-password-field is not provided. default_password_field = password +# Field/property containing the username, if --vault-username-field is not provided. +default_username_field = username + # 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 815b3508..38c32798 100644 --- a/test/pytests/test_cli_runner.py +++ b/test/pytests/test_cli_runner.py @@ -52,7 +52,7 @@ def default_config() -> dict[str, Any]: return { 'main': {'use_keyring': 'false'}, 'connection': {'default_keepalive_ticks': 0}, - 'vault': {}, + 'vault_beta': {}, 'alias_dsn': {}, 'init-commands': {}, 'alias_dsn.init-commands': {}, @@ -545,7 +545,8 @@ 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' + cli_args.user = 'existing-user' + cli_args.vault_secret = 'database/prod' client = DummyMyCli( config={ **default_config(), @@ -559,12 +560,12 @@ def test_run_from_cli_args_reads_password_from_vault_when_password_is_missing( ) vault_calls: list[dict[str, str | None]] = [] - def fake_get_password_from_vault(**kwargs: str | None) -> str: - vault_calls.append(kwargs) + def fake_get_field_from_vault(field: str, secret: str, **kwargs: str | None) -> str: + vault_calls.append({'field': field, 'secret': secret, **kwargs}) return 'vault-secret' monkeypatch.delenv('VAULT_ADDR', raising=False) - monkeypatch.setattr(cli_runner, 'get_password_from_vault', fake_get_password_from_vault) + monkeypatch.setattr(cli_runner, 'get_field_from_vault', fake_get_field_from_vault) run_with_client(monkeypatch, cli_args, client) @@ -580,18 +581,101 @@ def fake_get_password_from_vault(**kwargs: str | None) -> str: ] +def test_run_from_cli_args_reads_username_from_vault_when_user_is_missing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + cli_args = make_cli_args() + cli_args.password = 'existing-password' + cli_args.vault_secret = 'database/prod' + cli_args.vault_username_field = 'mysql_username' + client = DummyMyCli( + config={ + **default_config(), + 'vault_beta': { + 'vault_executable': '/opt/bin/vault', + 'address': 'https://vault.config', + 'default_mount': 'kv', + }, + } + ) + vault_calls: list[dict[str, str | None]] = [] + + def fake_get_field_from_vault(field: str, secret: str, **kwargs: str | None) -> str: + vault_calls.append({'field': field, 'secret': secret, **kwargs}) + return 'vault-user' + + monkeypatch.delenv('VAULT_ADDR', raising=False) + monkeypatch.setattr(cli_runner, 'get_field_from_vault', fake_get_field_from_vault) + + run_with_client(monkeypatch, cli_args, client) + + assert client.connect_calls[-1]['user'] == 'vault-user' + assert vault_calls == [ + { + 'secret': 'database/prod', + 'executable': '/opt/bin/vault', + 'field': 'mysql_username', + 'mount': 'kv', + 'address': 'https://vault.config', + } + ] + + +def test_run_from_cli_args_prefers_dsn_user_over_vault_username(monkeypatch: pytest.MonkeyPatch) -> None: + cli_args = make_cli_args() + cli_args.dsn = 'mysql://dsn-user@host/db' + cli_args.password = 'existing-password' + cli_args.vault_secret = 'database/prod' + cli_args.vault_username_field = 'mysql_username' + client = DummyMyCli() + vault_calls: list[dict[str, str | None]] = [] + + def fake_get_field_from_vault(field: str, secret: str, **kwargs: str | None) -> str: + vault_calls.append({'field': field, 'secret': secret, **kwargs}) + return 'vault-user' + + monkeypatch.setattr(cli_runner, 'get_field_from_vault', fake_get_field_from_vault) + + run_with_client(monkeypatch, cli_args, client) + + assert client.connect_calls[-1]['user'] == 'dsn-user' + assert vault_calls == [] + + +def test_run_from_cli_args_reports_vault_username_error(monkeypatch: pytest.MonkeyPatch) -> None: + cli_args = make_cli_args() + cli_args.password = 'existing-password' + cli_args.vault_secret = 'database/prod' + cli_args.vault_username_field = 'mysql_username' + 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_field_from_vault', + lambda *_args, **_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 username from Vault: permission denied', {'err': True, 'fg': 'red'})] + + 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' + cli_args.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) + def fake_get_field_from_vault(field: str, secret: str, **kwargs: str | None) -> str: + vault_calls.append({'field': field, 'secret': secret, **kwargs}) return 'vault-secret' - monkeypatch.setattr(cli_runner, 'get_password_from_vault', fake_get_password_from_vault) + monkeypatch.setattr(cli_runner, 'get_field_from_vault', fake_get_field_from_vault) run_with_client(monkeypatch, cli_args, client) @@ -603,9 +687,10 @@ 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' + cli_args.user = 'existing-user' + cli_args.vault_secret = 'database/prod' + cli_args.vault_mount = 'cli-mount' + cli_args.vault_password_field = 'cli-field' client = DummyMyCli( config={ **default_config(), @@ -620,11 +705,11 @@ def test_run_from_cli_args_prefers_vault_cli_values_and_env_address( 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) + def fake_get_field_from_vault(field: str, secret: str, **kwargs: str | None) -> str: + vault_calls.append({'field': field, 'secret': secret, **kwargs}) return 'vault-secret' - monkeypatch.setattr(cli_runner, 'get_password_from_vault', fake_get_password_from_vault) + monkeypatch.setattr(cli_runner, 'get_field_from_vault', fake_get_field_from_vault) run_with_client(monkeypatch, cli_args, client) @@ -644,17 +729,18 @@ 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' + cli_args.user = 'existing-user' + cli_args.vault_secret = 'database/prod' + cli_args.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) + def fake_get_field_from_vault(field: str, secret: str, **kwargs: str | None) -> str: + vault_calls.append({'field': field, 'secret': secret, **kwargs}) return 'vault-secret' - monkeypatch.setattr(cli_runner, 'get_password_from_vault', fake_get_password_from_vault) + monkeypatch.setattr(cli_runner, 'get_field_from_vault', fake_get_field_from_vault) run_with_client(monkeypatch, cli_args, client) @@ -664,14 +750,15 @@ def fake_get_password_from_vault(**kwargs: str | None) -> str: 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' + cli_args.user = 'existing-user' + cli_args.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')), + 'get_field_from_vault', + lambda *_args, **_kwargs: (_ for _ in ()).throw(cli_runner.VaultError('permission denied')), ) with pytest.raises(SystemExit) as excinfo: diff --git a/test/pytests/test_main.py b/test/pytests/test_main.py index e46c22d1..51c3a1a9 100644 --- a/test/pytests/test_main.py +++ b/test/pytests/test_main.py @@ -2392,22 +2392,25 @@ def test_click_entrypoint_populates_password_vault_options(monkeypatch: pytest.M result = CliRunner().invoke( click_entrypoint, [ - '--password-vault-address', + '--vault-address', 'https://vault.example.com', - '--password-vault-mount', + '--vault-mount', 'kv', - '--password-vault-secret', + '--vault-secret', 'database/prod', - '--password-vault-field', + '--vault-password-field', 'mysql_password', + '--vault-username-field', + 'mysql_username', ], ) 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' + assert cli_args_calls[-1].vault_address == 'https://vault.example.com' + assert cli_args_calls[-1].vault_mount == 'kv' + assert cli_args_calls[-1].vault_secret == 'database/prod' + assert cli_args_calls[-1].vault_password_field == 'mysql_password' + assert cli_args_calls[-1].vault_username_field == 'mysql_username' @pytest.mark.parametrize( diff --git a/test/pytests/test_vault.py b/test/pytests/test_vault.py index f3c3f398..5f11ec1c 100644 --- a/test/pytests/test_vault.py +++ b/test/pytests/test_vault.py @@ -9,7 +9,7 @@ from mycli import vault -def test_get_password_from_vault_runs_kv_get_with_field_mount_and_address( +def test_get_field_from_vault_runs_kv_get_with_field_mount_and_address( monkeypatch: pytest.MonkeyPatch, ) -> None: run_calls: list[dict[str, Any]] = [] @@ -20,10 +20,10 @@ def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace: monkeypatch.setattr(vault.subprocess, 'run', fake_run) - password = vault.get_password_from_vault( - secret='database/prod', + password = vault.get_field_from_vault( + 'mysql_password', + 'database/prod', executable='/opt/bin/vault', - field='mysql_password', mount='kv', address='https://vault.example.com', ) @@ -49,27 +49,27 @@ def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace: ] -def test_get_password_from_vault_reports_missing_executable(monkeypatch: pytest.MonkeyPatch) -> None: +def test_get_field_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') + vault.get_field_from_vault('password', 'database/prod', executable='missing-vault') -def test_get_password_from_vault_reports_oserror(monkeypatch: pytest.MonkeyPatch) -> None: +def test_get_field_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') + vault.get_field_from_vault('password', 'database/prod') -def test_get_password_from_vault_reports_nonzero_exit_without_stdout( +def test_get_field_from_vault_reports_nonzero_exit_without_stdout( monkeypatch: pytest.MonkeyPatch, ) -> None: def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace: @@ -78,13 +78,13 @@ def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace: monkeypatch.setattr(vault.subprocess, 'run', fake_run) with pytest.raises(vault.VaultError) as excinfo: - vault.get_password_from_vault(secret='database/prod') + vault.get_field_from_vault('password', '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( +def test_get_field_from_vault_reports_nonzero_exit_without_stderr( monkeypatch: pytest.MonkeyPatch, ) -> None: def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace: @@ -93,4 +93,4 @@ def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace: 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') + vault.get_field_from_vault('password', 'database/prod')