diff --git a/mycli/boundary_tunnel.py b/mycli/boundary_tunnel.py
index d16dbfcd..f70762b5 100644
--- a/mycli/boundary_tunnel.py
+++ b/mycli/boundary_tunnel.py
@@ -1,13 +1,19 @@
 from __future__ import annotations
 
+from collections.abc import Iterator
+from contextlib import ExitStack, contextmanager
 import datetime
 import json
 import os
 import shlex
 import socket
 import subprocess
+import sys
 import threading
 import time
+from typing import IO
+
+from mycli.compat import WIN
 
 
 class BoundaryTunnelError(RuntimeError):
@@ -20,6 +26,32 @@ def _find_free_local_port() -> int:
         return int(sock.getsockname()[1])
 
 
+@contextmanager
+def _authentication_terminal() -> Iterator[tuple[IO[str], IO[str]]]:
+    if sys.stdin.isatty() and sys.stderr.isatty():
+        yield sys.stdin, sys.stderr
+        return
+
+    with ExitStack() as stack:
+        try:
+            if WIN:
+                terminal_input = stack.enter_context(open('CONIN$', 'r', encoding='utf-8'))
+                terminal_output = stack.enter_context(open('CONOUT$', 'w', encoding='utf-8'))
+            else:
+                terminal_input = terminal_output = stack.enter_context(open('/dev/tty', 'r+', encoding='utf-8'))
+        except OSError as exc:
+            raise BoundaryTunnelError('Unable to open a terminal for Boundary authentication.') from exc
+        yield terminal_input, terminal_output
+
+
+def _prompt_for_authentication(terminal_input: IO[str], terminal_output: IO[str], prompt: str) -> str:
+    print(prompt, file=terminal_output, end='', flush=True)
+    response = terminal_input.readline()
+    if not response:
+        raise BoundaryTunnelError('Unable to read a response from the terminal.')
+    return response.rstrip('\r\n')
+
+
 class BoundaryTunnel:
     def __init__(
         self,
@@ -29,6 +61,8 @@ def __init__(
         address: str | None = None,
         auth_method_id: str | None = None,
         boundary_options: str | None = None,
+        boundary_test_command: str | None = None,
+        boundary_auth_command: str | None = None,
         local_port: int | None = None,
         ready_timeout: float = 30.0,
     ) -> None:
@@ -37,6 +71,8 @@ def __init__(
         self.address = address
         self.auth_method_id = auth_method_id
         self.boundary_options = boundary_options
+        self.boundary_test_command = boundary_test_command
+        self.boundary_auth_command = boundary_auth_command
         self.local_host = '127.0.0.1'
         self.local_port = local_port or _find_free_local_port()
         self.ready_timeout = ready_timeout
@@ -68,6 +104,7 @@ def command(self) -> list[str]:
         return command
 
     def start(self, *, show_expiration_warning: bool = True) -> None:
+        self._authenticate_if_needed()
         self._thread = threading.Thread(target=self._run, name='mycli-boundary-tunnel', daemon=True)
         self._thread.start()
         deadline = time.monotonic() + self.ready_timeout
@@ -78,7 +115,7 @@ def start(self, *, show_expiration_warning: bool = True) -> None:
             time.sleep(0.05)
         else:
             self.close()
-            raise BoundaryTunnelError('Timed out waiting for Boundary tunnel process to start.')
+            raise BoundaryTunnelError('Timed out waiting for tunnel process to start.')
 
         while time.monotonic() < deadline:
             self._raise_if_failed()
@@ -87,18 +124,18 @@ def start(self, *, show_expiration_warning: bool = True) -> None:
             time.sleep(0.05)
         else:
             self.close()
-            raise BoundaryTunnelError('Timed out waiting for Boundary tunnel process output.')
+            raise BoundaryTunnelError('Timed out waiting for tunnel process output.')
 
         connection_details = json.loads(self.stdout)
 
         if 'status_code' in connection_details:
-            raise BoundaryTunnelError(f'Boundary tunnel CLI raised status code {connection_details["status_code"]}.')
+            raise BoundaryTunnelError(f'Tunnel CLI raised status code {connection_details["status_code"]}.')
 
         try:
             self.username = connection_details['credentials'][0]['secret']['decoded']['username']
             self.password = connection_details['credentials'][0]['secret']['decoded']['password']
         except (IndexError, KeyError):
-            raise BoundaryTunnelError('Boundary tunnel CLI did not return credentials.') from None
+            raise BoundaryTunnelError('Tunnel CLI did not return credentials.') from None
 
         expiry_raw = connection_details['expiration']
         expiry_utc = datetime.datetime.strptime(expiry_raw, '%Y-%m-%dT%H:%M:%S.%f%z')
@@ -112,15 +149,15 @@ def start(self, *, show_expiration_warning: bool = True) -> None:
                 return
             time.sleep(0.05)
         self.close()
-        raise BoundaryTunnelError('Timed out waiting for Boundary tunnel to become ready.')
+        raise BoundaryTunnelError('Timed out waiting for tunnel to become ready.')
 
     def _raise_if_failed(self) -> None:
         if not self._failed.is_set():
             return
         self.close()
         if self._startup_error is not None:
-            raise BoundaryTunnelError(f'Unable to start Boundary tunnel process: {self._startup_error}') from self._startup_error
-        raise BoundaryTunnelError('Boundary tunnel process exited before it was ready.')
+            raise BoundaryTunnelError(f'Unable to start tunnel process: {self._startup_error}') from self._startup_error
+        raise BoundaryTunnelError('Tunnel process exited before it was ready.')
 
     def _read_stdout(self) -> None:
         process = self.process
@@ -129,12 +166,77 @@ def _read_stdout(self) -> None:
         self.stdout = process.stdout.readline().decode('utf-8')
 
     def _environment(self) -> dict[str, str] | None:
-        if not self.auth_method_id:
-            return None
         environment = os.environ.copy()
-        environment['BOUNDARY_AUTH_METHOD_ID'] = self.auth_method_id
+        if self.auth_method_id:
+            environment['BOUNDARY_AUTH_METHOD_ID'] = self.auth_method_id
+        if self.address:
+            environment['BOUNDARY_ADDR'] = self.address
         return environment
 
+    def _authenticate_if_needed(self) -> None:
+        if not self.boundary_test_command or not self.boundary_auth_command:
+            return
+
+        test_command = self._parse_authentication_command(self.boundary_test_command, 'test')
+        try:
+            completed_process = subprocess.run(
+                test_command,
+                check=False,
+                stdin=subprocess.DEVNULL,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                env=self._environment(),
+            )
+        except OSError as exc:
+            raise BoundaryTunnelError(f'Unable to run test command: {exc}') from exc
+        if completed_process.returncode == 0:
+            return
+
+        with _authentication_terminal() as (terminal_input, terminal_output):
+            yn = _prompt_for_authentication(
+                terminal_input,
+                terminal_output,
+                'Authenticate with Boundary before connecting? [Yn] ',
+            ).lower()
+            if yn not in ('y', ''):
+                raise BoundaryTunnelError('Not authenticated.')
+
+            auth_command = self._parse_authentication_command(self.boundary_auth_command, 'authentication')
+            try:
+                completed_process = subprocess.run(
+                    auth_command,
+                    check=False,
+                    stdin=terminal_input,
+                    stdout=terminal_output,
+                    stderr=terminal_output,
+                    env=self._environment(),
+                )
+            except OSError as exc:
+                raise BoundaryTunnelError(f'Unable to run authentication command: {exc}') from exc
+            if completed_process.returncode != 0:
+                raise BoundaryTunnelError(f'Authentication command exited with status {completed_process.returncode}.')
+
+            _prompt_for_authentication(
+                terminal_input,
+                terminal_output,
+                'Press return to continue after authenticating: ',
+            )
+
+    @staticmethod
+    def _parse_authentication_command(command: str, name: str) -> list[str]:
+        try:
+            arguments = shlex.split(command, posix=not WIN)
+        except ValueError as exc:
+            raise BoundaryTunnelError(f'Unable to parse {name} command: {exc}') from exc
+        if WIN:
+            arguments = [
+                argument[1:-1] if len(argument) >= 2 and argument[0] == argument[-1] and argument[0] in ('"', "'") else argument
+                for argument in arguments
+            ]
+        if not arguments:
+            raise BoundaryTunnelError(f'{name} command is empty.')
+        return arguments
+
     def close(self) -> None:
         process = self.process
         if process is not None and process.poll() is None:
diff --git a/mycli/client_connection.py b/mycli/client_connection.py
index 16d3c4e0..900c9d0e 100644
--- a/mycli/client_connection.py
+++ b/mycli/client_connection.py
@@ -136,6 +136,8 @@ def connect(
             boundary_address = self.config.get('boundary_beta', {}).get('address') or None
             boundary_auth_method_id = self.config.get('boundary_beta', {}).get('auth_method_id') or None
             boundary_options = self.config.get('boundary_beta', {}).get('boundary_options') or None
+            boundary_test_command = self.config.get('boundary_beta', {}).get('boundary_test_command') or None
+            boundary_auth_command = self.config.get('boundary_beta', {}).get('boundary_auth_command') or None
             try:
                 self.boundary_tunnel = BoundaryTunnel(
                     target_id=boundary_target_id,
@@ -143,6 +145,8 @@ def connect(
                     address=boundary_address,
                     auth_method_id=boundary_auth_method_id,
                     boundary_options=boundary_options,
+                    boundary_test_command=boundary_test_command,
+                    boundary_auth_command=boundary_auth_command,
                 )
                 self.boundary_tunnel.start(show_expiration_warning=self.verbosity >= 0)
                 if self.verbosity >= 0:
diff --git a/mycli/myclirc b/mycli/myclirc
index e1545849..4c750e79 100644
--- a/mycli/myclirc
+++ b/mycli/myclirc
@@ -435,6 +435,18 @@ auth_method_id =
 # Additional options passed to the "boundary connect" command.
 boundary_options =
 
+# Command used to test whether Boundary authentication is valid.
+# If empty, establishing the tunnel will fail when not authenticated.
+# If set, boundary_auth_command must also be set.
+# Example: boundary targets read -id ttcp_zzzzzzzzzz
+boundary_test_command =
+
+# Command used to authenticate when boundary_test_command exits nonzero.
+# If empty, establishing the tunnel will fail when not authenticated.
+# If set, boundary_test_command must also be set.
+# Example: boundary authenticate oidc
+boundary_auth_command =
+
 # 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/myclirc b/test/myclirc
index cbaf13a8..a4bcd270 100644
--- a/test/myclirc
+++ b/test/myclirc
@@ -435,6 +435,18 @@ auth_method_id =
 # Additional options passed to the "boundary connect" command.
 boundary_options =
 
+# Command used to test whether Boundary authentication is valid.
+# If empty, establishing the tunnel will fail when not authenticated.
+# If set, boundary_auth_command must also be set.
+# Example: boundary targets read -id ttcp_zzzzzzzzzz
+boundary_test_command =
+
+# Command used to authenticate when boundary_test_command exits nonzero.
+# If empty, establishing the tunnel will fail when not authenticated.
+# If set, boundary_test_command must also be set.
+# Example: boundary authenticate oidc
+boundary_auth_command =
+
 # 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_boundary_tunnel.py b/test/pytests/test_boundary_tunnel.py
index fcb7cc64..0aaed81b 100644
--- a/test/pytests/test_boundary_tunnel.py
+++ b/test/pytests/test_boundary_tunnel.py
@@ -1,9 +1,12 @@
 from __future__ import annotations
 
+from contextlib import nullcontext
+from io import StringIO
 import os
 import socket
 import subprocess
 import threading
+from types import SimpleNamespace
 from typing import Any, cast
 
 import pytest
@@ -72,7 +75,10 @@ def test_boundary_tunnel_command_splits_options_before_generated_flags() -> None
 def test_boundary_tunnel_environment_uses_parent_default(auth_method_id: str | None) -> None:
     tunnel = BoundaryTunnel(target_id='ttcp_123', auth_method_id=auth_method_id)
 
-    assert tunnel._environment() is None
+    environment = tunnel._environment()
+
+    assert environment is not None
+    assert environment == dict(os.environ)
 
 
 def test_boundary_tunnel_environment_sets_configured_auth_method(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -86,6 +92,347 @@ def test_boundary_tunnel_environment_sets_configured_auth_method(monkeypatch: py
     assert os.environ['BOUNDARY_AUTH_METHOD_ID'] == 'ampw_parent'
 
 
+def test_boundary_tunnel_environment_sets_configured_address(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv('BOUNDARY_ADDR', 'https://parent.example.com')
+    tunnel = BoundaryTunnel(target_id='ttcp_123', address='https://config.example.com')
+
+    environment = tunnel._environment()
+
+    assert environment is not None
+    assert environment['BOUNDARY_ADDR'] == 'https://config.example.com'
+    assert os.environ['BOUNDARY_ADDR'] == 'https://parent.example.com'
+
+
+@pytest.mark.parametrize(
+    ('test_command', 'auth_command'),
+    [
+        (None, 'boundary authenticate'),
+        ('boundary authenticate status', None),
+        ('', 'boundary authenticate'),
+        ('boundary authenticate status', ''),
+    ],
+)
+def test_boundary_tunnel_authentication_requires_both_commands(
+    monkeypatch: pytest.MonkeyPatch,
+    test_command: str | None,
+    auth_command: str | None,
+) -> None:
+    monkeypatch.setattr(boundary_tunnel.subprocess, 'run', lambda *_args, **_kwargs: pytest.fail('unexpected command'))
+    tunnel = BoundaryTunnel(
+        target_id='ttcp_123',
+        boundary_test_command=test_command,
+        boundary_auth_command=auth_command,
+        local_port=4406,
+    )
+
+    tunnel._authenticate_if_needed()
+
+
+def test_boundary_tunnel_authentication_skips_auth_when_test_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: list[tuple[list[str], dict[str, Any]]] = []
+
+    def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace:
+        calls.append((command, kwargs))
+        return SimpleNamespace(returncode=0)
+
+    monkeypatch.setattr(boundary_tunnel.subprocess, 'run', fake_run)
+    tunnel = BoundaryTunnel(
+        target_id='ttcp_123',
+        auth_method_id='ampw_123',
+        boundary_test_command='boundary authenticate status -name "my auth"',
+        boundary_auth_command='boundary authenticate password',
+        local_port=4406,
+    )
+
+    tunnel._authenticate_if_needed()
+
+    assert calls == [
+        (
+            ['boundary', 'authenticate', 'status', '-name', 'my auth'],
+            {
+                'check': False,
+                'stdin': subprocess.DEVNULL,
+                'stdout': subprocess.DEVNULL,
+                'stderr': subprocess.DEVNULL,
+                'env': {**os.environ, 'BOUNDARY_AUTH_METHOD_ID': 'ampw_123'},
+            },
+        )
+    ]
+
+
+def test_boundary_tunnel_authentication_runs_auth_after_failed_test(monkeypatch: pytest.MonkeyPatch) -> None:
+    class FakeTerminal(StringIO):
+        def __init__(self, responses: str) -> None:
+            super().__init__(responses)
+            self.output = StringIO()
+
+        def write(self, value: str) -> int:
+            return self.output.write(value)
+
+        def flush(self) -> None:
+            self.output.flush()
+
+        def close(self) -> None:
+            pass
+
+    calls: list[tuple[list[str], dict[str, Any]]] = []
+    return_codes = iter([1, 0])
+    sql_input = StringIO('select 1;\n')
+    redirected_error = StringIO()
+    terminal = FakeTerminal('\n\n')
+    opened_paths: list[tuple[str, str, str]] = []
+
+    def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace:
+        calls.append((command, kwargs))
+        return SimpleNamespace(returncode=next(return_codes))
+
+    def fake_open(path: str, mode: str, *, encoding: str) -> FakeTerminal:
+        opened_paths.append((path, mode, encoding))
+        return terminal
+
+    monkeypatch.setattr(boundary_tunnel.subprocess, 'run', fake_run)
+    monkeypatch.setattr(boundary_tunnel, 'WIN', False)
+    monkeypatch.setattr(boundary_tunnel.sys, 'stdin', sql_input)
+    monkeypatch.setattr(boundary_tunnel.sys, 'stderr', redirected_error)
+    monkeypatch.setattr(boundary_tunnel, 'open', fake_open, raising=False)
+    tunnel = BoundaryTunnel(
+        target_id='ttcp_123',
+        boundary_test_command='boundary authenticate status',
+        boundary_auth_command='boundary authenticate password -login-name "Jane Doe"',
+        local_port=4406,
+    )
+
+    tunnel._authenticate_if_needed()
+
+    assert calls == [
+        (
+            ['boundary', 'authenticate', 'status'],
+            {
+                'check': False,
+                'stdin': subprocess.DEVNULL,
+                'stdout': subprocess.DEVNULL,
+                'stderr': subprocess.DEVNULL,
+                'env': {**os.environ},
+            },
+        ),
+        (
+            ['boundary', 'authenticate', 'password', '-login-name', 'Jane Doe'],
+            {
+                'check': False,
+                'stdin': terminal,
+                'stdout': terminal,
+                'stderr': terminal,
+                'env': {**os.environ},
+            },
+        ),
+    ]
+    assert opened_paths == [('/dev/tty', 'r+', 'utf-8')]
+    assert sql_input.read() == 'select 1;\n'
+    assert redirected_error.getvalue() == ''
+    assert terminal.output.getvalue() == (
+        'Authenticate with Boundary before connecting? [Yn] Press return to continue after authenticating: '
+    )
+
+
+def test_boundary_tunnel_authentication_uses_tty_streams(monkeypatch: pytest.MonkeyPatch) -> None:
+    terminal_input = StringIO('response\n')
+    terminal_output = StringIO()
+    monkeypatch.setattr(terminal_input, 'isatty', lambda: True)
+    monkeypatch.setattr(terminal_output, 'isatty', lambda: True)
+    monkeypatch.setattr(boundary_tunnel.sys, 'stdin', terminal_input)
+    monkeypatch.setattr(boundary_tunnel.sys, 'stderr', terminal_output)
+
+    with boundary_tunnel._authentication_terminal() as (authentication_input, authentication_output):
+        assert authentication_input is terminal_input
+        assert authentication_output is terminal_output
+
+
+def test_boundary_tunnel_authentication_opens_windows_console(monkeypatch: pytest.MonkeyPatch) -> None:
+    terminal_input = StringIO('response\n')
+    terminal_output = StringIO()
+    opened_paths: list[tuple[str, str, str]] = []
+
+    def fake_open(path: str, mode: str, *, encoding: str) -> StringIO:
+        opened_paths.append((path, mode, encoding))
+        return terminal_input if path == 'CONIN$' else terminal_output
+
+    monkeypatch.setattr(boundary_tunnel, 'WIN', True)
+    monkeypatch.setattr(boundary_tunnel.sys, 'stdin', StringIO())
+    monkeypatch.setattr(boundary_tunnel.sys, 'stderr', StringIO())
+    monkeypatch.setattr(boundary_tunnel, 'open', fake_open, raising=False)
+
+    with boundary_tunnel._authentication_terminal() as (authentication_input, authentication_output):
+        assert authentication_input is terminal_input
+        assert authentication_output is terminal_output
+
+    assert opened_paths == [
+        ('CONIN$', 'r', 'utf-8'),
+        ('CONOUT$', 'w', 'utf-8'),
+    ]
+
+
+def test_boundary_tunnel_authentication_reports_missing_terminal(monkeypatch: pytest.MonkeyPatch) -> None:
+    sql_input = StringIO('select 1;\n')
+
+    def fail_open(*_args: Any, **_kwargs: Any) -> None:
+        raise OSError('no terminal')
+
+    monkeypatch.setattr(boundary_tunnel.sys, 'stdin', sql_input)
+    monkeypatch.setattr(boundary_tunnel.sys, 'stderr', StringIO())
+    monkeypatch.setattr(boundary_tunnel, 'open', fail_open, raising=False)
+
+    with pytest.raises(BoundaryTunnelError, match='Unable to open a terminal') as excinfo:
+        with boundary_tunnel._authentication_terminal():
+            pass
+
+    assert isinstance(excinfo.value.__cause__, OSError)
+    assert sql_input.read() == 'select 1;\n'
+
+
+def test_boundary_tunnel_authentication_reports_terminal_eof() -> None:
+    with pytest.raises(BoundaryTunnelError, match='Unable to read a response from the terminal'):
+        boundary_tunnel._prompt_for_authentication(StringIO(), StringIO(), 'Prompt: ')
+
+
+def test_boundary_tunnel_authentication_reports_declined_auth(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        boundary_tunnel.subprocess,
+        'run',
+        lambda *_args, **_kwargs: SimpleNamespace(returncode=1),
+    )
+    monkeypatch.setattr(
+        boundary_tunnel,
+        '_authentication_terminal',
+        lambda: nullcontext((StringIO('n\n'), StringIO())),
+    )
+    tunnel = BoundaryTunnel(
+        target_id='ttcp_123',
+        boundary_test_command='boundary authenticate status',
+        boundary_auth_command='boundary authenticate password',
+        local_port=4406,
+    )
+
+    with pytest.raises(BoundaryTunnelError, match='Not authenticated'):
+        tunnel._authenticate_if_needed()
+
+
+@pytest.mark.parametrize(
+    ('command', 'expected'),
+    [
+        (
+            r'C:\boundary\boundary.exe authenticate status',
+            [r'C:\boundary\boundary.exe', 'authenticate', 'status'],
+        ),
+        (
+            r'"C:\Program Files\Boundary\boundary.exe" authenticate -name "Jane Doe"',
+            [r'C:\Program Files\Boundary\boundary.exe', 'authenticate', '-name', 'Jane Doe'],
+        ),
+    ],
+)
+def test_boundary_tunnel_authentication_parses_windows_commands(
+    monkeypatch: pytest.MonkeyPatch,
+    command: str,
+    expected: list[str],
+) -> None:
+    monkeypatch.setattr(boundary_tunnel, 'WIN', True)
+
+    assert BoundaryTunnel._parse_authentication_command(command, 'test') == expected
+
+
+def test_boundary_tunnel_authentication_reports_failed_auth(monkeypatch: pytest.MonkeyPatch) -> None:
+    return_codes = iter([1, 2])
+    monkeypatch.setattr(
+        boundary_tunnel.subprocess,
+        'run',
+        lambda *_args, **_kwargs: SimpleNamespace(returncode=next(return_codes)),
+    )
+    monkeypatch.setattr(
+        boundary_tunnel,
+        '_authentication_terminal',
+        lambda: nullcontext((StringIO('\n'), StringIO())),
+    )
+    tunnel = BoundaryTunnel(
+        target_id='ttcp_123',
+        boundary_test_command='boundary authenticate status',
+        boundary_auth_command='boundary authenticate password',
+        local_port=4406,
+    )
+
+    with pytest.raises(BoundaryTunnelError, match='Authentication command exited with status 2'):
+        tunnel._authenticate_if_needed()
+
+
+@pytest.mark.parametrize(
+    ('test_command', 'auth_command', 'error_match'),
+    [
+        ('"unterminated', 'boundary authenticate', 'Unable to parse test command'),
+        ('   ', 'boundary authenticate', 'test command is empty'),
+        ('boundary test', '"unterminated', 'Unable to parse authentication command'),
+        ('boundary test', '   ', 'authentication command is empty'),
+    ],
+)
+def test_boundary_tunnel_authentication_reports_invalid_commands(
+    monkeypatch: pytest.MonkeyPatch,
+    test_command: str,
+    auth_command: str,
+    error_match: str,
+) -> None:
+    monkeypatch.setattr(boundary_tunnel.subprocess, 'run', lambda *_args, **_kwargs: SimpleNamespace(returncode=1))
+    monkeypatch.setattr(
+        boundary_tunnel,
+        '_authentication_terminal',
+        lambda: nullcontext((StringIO('\n'), StringIO())),
+    )
+    tunnel = BoundaryTunnel(
+        target_id='ttcp_123',
+        boundary_test_command=test_command,
+        boundary_auth_command=auth_command,
+        local_port=4406,
+    )
+
+    with pytest.raises(BoundaryTunnelError, match=error_match):
+        tunnel._authenticate_if_needed()
+
+
+@pytest.mark.parametrize(
+    ('return_codes', 'error_match'),
+    [
+        ([], 'Unable to run test command: command failed'),
+        ([1], 'Unable to run authentication command: command failed'),
+    ],
+)
+def test_boundary_tunnel_authentication_reports_process_errors(
+    monkeypatch: pytest.MonkeyPatch,
+    return_codes: list[int],
+    error_match: str,
+) -> None:
+    remaining_codes = iter(return_codes)
+
+    def fake_run(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
+        try:
+            return SimpleNamespace(returncode=next(remaining_codes))
+        except StopIteration:
+            raise OSError('command failed') from None
+
+    monkeypatch.setattr(boundary_tunnel.subprocess, 'run', fake_run)
+    monkeypatch.setattr(
+        boundary_tunnel,
+        '_authentication_terminal',
+        lambda: nullcontext((StringIO('\n'), StringIO())),
+    )
+    tunnel = BoundaryTunnel(
+        target_id='ttcp_123',
+        boundary_test_command='boundary authenticate status',
+        boundary_auth_command='boundary authenticate password',
+        local_port=4406,
+    )
+
+    with pytest.raises(BoundaryTunnelError, match=error_match) as excinfo:
+        tunnel._authenticate_if_needed()
+
+    assert isinstance(excinfo.value.__cause__, OSError)
+
+
 def test_boundary_tunnel_allocates_local_port(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(boundary_tunnel, '_find_free_local_port', lambda: 4406)
 
@@ -141,6 +488,28 @@ def fake_read_stdout() -> None:
     assert tunnel.expiry is not None
 
 
+def test_boundary_tunnel_start_authenticates_before_starting_worker(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: list[str] = []
+    tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406)
+    tunnel.stdout = CONNECTION_DETAILS
+
+    def fake_authenticate() -> None:
+        calls.append('authenticate')
+
+    def fake_run() -> None:
+        calls.append('run')
+        tunnel._started.set()
+        tunnel._output_ready.set()
+
+    monkeypatch.setattr(tunnel, '_authenticate_if_needed', fake_authenticate)
+    monkeypatch.setattr(tunnel, '_run', fake_run)
+    monkeypatch.setattr(tunnel, '_is_listening', lambda: True)
+
+    tunnel.start()
+
+    assert calls == ['authenticate', 'run']
+
+
 def test_boundary_tunnel_start_waits_for_process_to_start(monkeypatch: pytest.MonkeyPatch) -> None:
     sleeps: list[float] = []
     tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406)
@@ -170,7 +539,7 @@ def fake_run() -> None:
 
     monkeypatch.setattr(tunnel, '_run', fake_run)
 
-    with pytest.raises(BoundaryTunnelError, match='Boundary tunnel CLI raised status code 403'):
+    with pytest.raises(BoundaryTunnelError, match='Tunnel CLI raised status code 403'):
         tunnel.start()
 
 
@@ -191,7 +560,7 @@ def fake_run() -> None:
 
     monkeypatch.setattr(tunnel, '_run', fake_run)
 
-    with pytest.raises(BoundaryTunnelError, match='Boundary tunnel CLI did not return credentials'):
+    with pytest.raises(BoundaryTunnelError, match='Tunnel CLI did not return credentials'):
         tunnel.start()
 
 
@@ -254,7 +623,7 @@ def fake_read_stdout() -> None:
     monkeypatch.setattr(tunnel, '_run', fake_run)
     monkeypatch.setattr(tunnel, '_read_stdout', fake_read_stdout)
 
-    with pytest.raises(BoundaryTunnelError, match='Unable to start Boundary tunnel process: boundary failed') as excinfo:
+    with pytest.raises(BoundaryTunnelError, match='Unable to start tunnel process: boundary failed') as excinfo:
         tunnel.start()
 
     assert isinstance(excinfo.value.__cause__, OSError)
@@ -268,7 +637,7 @@ def fail_popen(*_args: Any, **_kwargs: Any) -> None:
     tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406)
     monkeypatch.setattr(tunnel, '_is_listening', lambda: False)
 
-    with pytest.raises(BoundaryTunnelError, match='Unable to start Boundary tunnel process: missing boundary') as excinfo:
+    with pytest.raises(BoundaryTunnelError, match='Unable to start tunnel process: missing boundary') as excinfo:
         tunnel.start()
 
     assert isinstance(excinfo.value.__cause__, FileNotFoundError)
@@ -277,7 +646,7 @@ def fail_popen(*_args: Any, **_kwargs: Any) -> None:
 def test_boundary_tunnel_start_reports_invalid_options() -> None:
     tunnel = BoundaryTunnel(target_id='ttcp_123', boundary_options='"unterminated', local_port=4406)
 
-    with pytest.raises(BoundaryTunnelError, match='Unable to start Boundary tunnel process: No closing quotation') as excinfo:
+    with pytest.raises(BoundaryTunnelError, match='Unable to start tunnel process: No closing quotation') as excinfo:
         tunnel.start()
 
     assert isinstance(excinfo.value.__cause__, ValueError)
@@ -287,7 +656,7 @@ def test_boundary_tunnel_start_reports_process_start_timeout(monkeypatch: pytest
     tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406, ready_timeout=0)
     monkeypatch.setattr(tunnel, '_run', lambda: None)
 
-    with pytest.raises(BoundaryTunnelError, match='Timed out waiting for Boundary tunnel process to start'):
+    with pytest.raises(BoundaryTunnelError, match='Timed out waiting for tunnel process to start'):
         tunnel.start()
 
 
@@ -318,7 +687,7 @@ def wait(self, timeout: float | None = None) -> int:
     monkeypatch.setattr(boundary_tunnel.subprocess, 'Popen', lambda *_args, **_kwargs: FakeProcess())
     tunnel = BoundaryTunnel(target_id='ttcp_123', local_port=4406, ready_timeout=0.1)
 
-    with pytest.raises(BoundaryTunnelError, match='Timed out waiting for Boundary tunnel process output'):
+    with pytest.raises(BoundaryTunnelError, match='Timed out waiting for tunnel process output'):
         tunnel.start()
 
     assert calls[:2] == ['read', 'terminate']
@@ -338,7 +707,7 @@ def fake_run() -> None:
     monotonic_values = iter([0.0, 0.0, 0.0, 31.0])
     monkeypatch.setattr(boundary_tunnel.time, 'monotonic', lambda: next(monotonic_values))
 
-    with pytest.raises(BoundaryTunnelError, match='Timed out waiting for Boundary tunnel'):
+    with pytest.raises(BoundaryTunnelError, match='Timed out waiting for tunnel'):
         tunnel.start()
 
 
@@ -411,7 +780,7 @@ def fake_popen(command: list[str], **kwargs: Any) -> FakeProcess:
                 'stdout': subprocess.PIPE,
                 'stderr': subprocess.DEVNULL,
                 'start_new_session': True,
-                'env': None,
+                'env': {**os.environ},
             },
         )
     ]
diff --git a/test/pytests/test_client_connection.py b/test/pytests/test_client_connection.py
index bcc3464e..e149dcb5 100644
--- a/test/pytests/test_client_connection.py
+++ b/test/pytests/test_client_connection.py
@@ -716,6 +716,8 @@ def __init__(
             address: str | None,
             auth_method_id: str | None,
             boundary_options: str | None,
+            boundary_test_command: str | None,
+            boundary_auth_command: str | None,
         ) -> None:
             tunnel_calls.append({
                 'target_id': target_id,
@@ -723,6 +725,8 @@ def __init__(
                 'address': address,
                 'auth_method_id': auth_method_id,
                 'boundary_options': boundary_options,
+                'boundary_test_command': boundary_test_command,
+                'boundary_auth_command': boundary_auth_command,
             })
 
         def start(self, *, show_expiration_warning: bool) -> None:
@@ -741,6 +745,8 @@ def close(self) -> None:
                 'address': 'https://boundary.example.com',
                 'auth_method_id': 'ampw_123',
                 'boundary_options': '-token env://BOUNDARY_TOKEN',
+                'boundary_test_command': 'boundary authenticate status',
+                'boundary_auth_command': 'boundary authenticate password',
             },
             'connection': {},
         },
@@ -762,6 +768,8 @@ def close(self) -> None:
             'address': 'https://boundary.example.com',
             'auth_method_id': 'ampw_123',
             'boundary_options': '-token env://BOUNDARY_TOKEN',
+            'boundary_test_command': 'boundary authenticate status',
+            'boundary_auth_command': 'boundary authenticate password',
         }
     ]
     assert warning_calls == [expected_warning]
@@ -804,6 +812,8 @@ def __init__(
             address: str | None,
             auth_method_id: str | None,
             boundary_options: str | None,
+            boundary_test_command: str | None,
+            boundary_auth_command: str | None,
         ) -> None:
             pass
 
@@ -834,7 +844,7 @@ def close(self) -> None:
 
 
 def test_connect_boundary_tunnel_uses_default_config(monkeypatch: pytest.MonkeyPatch) -> None:
-    tunnel_calls: list[tuple[str, str | None, str | None, str | None]] = []
+    tunnel_calls: list[tuple[str, str | None, str | None, str | None, str | None, str | None]] = []
     warning_calls: list[bool] = []
 
     class FakeTunnel:
@@ -852,8 +862,17 @@ def __init__(
             address: str | None,
             auth_method_id: str | None,
             boundary_options: str | None,
+            boundary_test_command: str | None,
+            boundary_auth_command: str | None,
         ) -> None:
-            tunnel_calls.append((boundary_executable, address, auth_method_id, boundary_options))
+            tunnel_calls.append((
+                boundary_executable,
+                address,
+                auth_method_id,
+                boundary_options,
+                boundary_test_command,
+                boundary_auth_command,
+            ))
 
         def start(self, *, show_expiration_warning: bool) -> None:
             warning_calls.append(show_expiration_warning)
@@ -866,7 +885,7 @@ def close(self) -> None:
 
     client.connect(host='db.internal', boundary_target_id='ttcp_123')
 
-    assert tunnel_calls == [('boundary', None, None, None)]
+    assert tunnel_calls == [('boundary', None, None, None, None, None)]
     assert warning_calls == [True]
 
 
@@ -887,6 +906,8 @@ def __init__(
             address: str | None,
             auth_method_id: str | None,
             boundary_options: str | None,
+            boundary_test_command: str | None,
+            boundary_auth_command: str | None,
         ) -> None:
             pass
 
@@ -930,6 +951,8 @@ def __init__(
             address: str | None,
             auth_method_id: str | None,
             boundary_options: str | None,
+            boundary_test_command: str | None,
+            boundary_auth_command: str | None,
         ) -> None:
             pass
 
@@ -967,6 +990,8 @@ def __init__(
             address: str | None,
             auth_method_id: str | None,
             boundary_options: str | None,
+            boundary_test_command: str | None,
+            boundary_auth_command: str | None,
         ) -> None:
             pass