Ensure user is logged in to Vault

changelog.md

@@ -4,6 +4,7 @@ Upcoming (TBD)
 Features
 ---------
 * Always clean favorite queries on save and fetch.
+* Give a clearer message on a Vault connection if the user is not logged in.
 
 
 Bugfixes

mycli/vault.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import functools
 import subprocess
 
 DEFAULT_VAULT_EXECUTABLE = 'vault'
@@ -11,13 +12,53 @@ class VaultError(RuntimeError):
     pass
 
 
+@functools.lru_cache(maxsize=32)
+def _ensure_vault_user_logged_in(
+    executable: str = DEFAULT_VAULT_EXECUTABLE,
+    address: str | None = None,
+) -> None:
+    command = [
+        executable,
+        'token',
+        'lookup',
+        '-format=json',
+    ]
+    if address:
+        command.append(f'-address={address}')
+
+    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:
+        # maybe could display something from the JSON output
+        # maybe only with --verbose
+        raise VaultError('Not logged in to Vault. You may need to run "vault login".')
+
+
 def get_field_from_vault(
     field: str,
     secret: str,
     executable: str = DEFAULT_VAULT_EXECUTABLE,
     mount: str | None = None,
     address: str | None = None,
 ) -> str:
+
+    _ensure_vault_user_logged_in(
+        executable=executable,
+        address=address,
+    )
+
     command = [
         executable,
         'kv',

test/pytests/test_vault.py

@@ -9,6 +9,11 @@
 from mycli import vault
 
 
+@pytest.fixture(autouse=True)
+def clear_vault_login_cache() -> None:
+    vault._ensure_vault_user_logged_in.cache_clear()
+
+
 def test_get_field_from_vault_runs_kv_get_with_field_mount_and_address(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -30,6 +35,20 @@ def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace:
 
     assert password == 'secret'
     assert run_calls == [
+        {
+            'command': [
+                '/opt/bin/vault',
+                'token',
+                'lookup',
+                '-format=json',
+                '-address=https://vault.example.com',
+            ],
+            'check': False,
+            'stdin': -3,
+            'stdout': -1,
+            'stderr': -1,
+            'text': True,
+        },
         {
             'command': [
                 '/opt/bin/vault',
@@ -45,7 +64,7 @@ def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace:
             'stdout': subprocess.PIPE,
             'stderr': subprocess.PIPE,
             'text': True,
-        }
+        },
     ]
 
 
@@ -72,7 +91,9 @@ def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
 def test_get_field_from_vault_reports_nonzero_exit_without_stdout(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
+    def fake_run(command: list[str], **_kwargs: Any) -> SimpleNamespace:
+        if command[1:3] == ['token', 'lookup']:
+            return SimpleNamespace(returncode=0, stdout='token metadata\n', stderr='')
         return SimpleNamespace(returncode=2, stdout='secret\n', stderr='permission denied\n')
 
     monkeypatch.setattr(vault.subprocess, 'run', fake_run)
@@ -84,6 +105,43 @@ def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
     assert 'secret' not in str(excinfo.value)
 
 
+@pytest.mark.parametrize(
+    ('error', 'message'),
+    (
+        (FileNotFoundError(), 'Vault executable not found: custom-vault'),
+        (OSError('boom'), 'Unable to run Vault executable custom-vault: boom'),
+    ),
+)
+def test_get_field_from_vault_reports_kv_get_start_error(
+    monkeypatch: pytest.MonkeyPatch,
+    error: OSError,
+    message: str,
+) -> None:
+    def fake_run(command: list[str], **_kwargs: Any) -> SimpleNamespace:
+        if command[1:3] == ['token', 'lookup']:
+            return SimpleNamespace(returncode=0, stdout='token metadata\n', stderr='')
+        raise error
+
+    monkeypatch.setattr(vault.subprocess, 'run', fake_run)
+
+    with pytest.raises(vault.VaultError, match=message):
+        vault.get_field_from_vault('password', 'database/prod', executable='custom-vault')
+
+
+def test_get_field_from_vault_reports_kv_get_nonzero_exit_without_stderr(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def fake_run(command: list[str], **_kwargs: Any) -> SimpleNamespace:
+        if command[1:3] == ['token', 'lookup']:
+            return SimpleNamespace(returncode=0, stdout='token metadata\n', stderr='')
+        return SimpleNamespace(returncode=2, stdout='', stderr='')
+
+    monkeypatch.setattr(vault.subprocess, 'run', fake_run)
+
+    with pytest.raises(vault.VaultError, match='Vault command failed.*Exit code 2'):
+        vault.get_field_from_vault('password', 'database/prod')
+
+
 def test_get_field_from_vault_reports_nonzero_exit_without_stderr(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -92,5 +150,5 @@ 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\\.'):
+    with pytest.raises(vault.VaultError, match='Not logged in to Vault. You may need to run "vault login".'):
         vault.get_field_from_vault('password', 'database/prod')