Allow taking username from Vault

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

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,

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

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.

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:

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.

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: